Merge commit 'vudentz/master'
[profile/ivi/pulseaudio-panda.git] / src / modules / module-raop-sink.c
1 /***
2   This file is part of PulseAudio.
3
4   Copyright 2004-2006 Lennart Poettering
5   Copyright 2008 Colin Guthrie
6
7   PulseAudio is free software; you can redistribute it and/or modify
8   it under the terms of the GNU Lesser General Public License as published
9   by the Free Software Foundation; either version 2 of the License,
10   or (at your option) any later version.
11
12   PulseAudio is distributed in the hope that it will be useful, but
13   WITHOUT ANY WARRANTY; without even the implied warranty of
14   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15   General Public License for more details.
16
17   You should have received a copy of the GNU Lesser General Public License
18   along with PulseAudio; if not, write to the Free Software
19   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
20   USA.
21 ***/
22
23 #ifdef HAVE_CONFIG_H
24 #include <config.h>
25 #endif
26
27 #include <stdlib.h>
28 #include <sys/stat.h>
29 #include <stdio.h>
30 #include <errno.h>
31 #include <string.h>
32 #include <fcntl.h>
33 #include <unistd.h>
34 #include <limits.h>
35 #include <poll.h>
36 #include <sys/socket.h>
37 #include <netinet/in.h>
38 #include <netinet/tcp.h>
39 #include <sys/ioctl.h>
40
41 #ifdef HAVE_LINUX_SOCKIOS_H
42 #include <linux/sockios.h>
43 #endif
44
45 #include <pulse/xmalloc.h>
46 #include <pulse/timeval.h>
47
48 #include <pulsecore/core-error.h>
49 #include <pulsecore/iochannel.h>
50 #include <pulsecore/sink.h>
51 #include <pulsecore/module.h>
52 #include <pulsecore/core-util.h>
53 #include <pulsecore/modargs.h>
54 #include <pulsecore/log.h>
55 #include <pulsecore/socket-client.h>
56 #include <pulsecore/authkey.h>
57 #include <pulsecore/thread-mq.h>
58 #include <pulsecore/thread.h>
59 #include <pulsecore/time-smoother.h>
60 #include <pulsecore/rtclock.h>
61 #include <pulsecore/socket-util.h>
62
63 #include "module-raop-sink-symdef.h"
64 #include "rtp.h"
65 #include "sdp.h"
66 #include "sap.h"
67 #include "raop_client.h"
68
69 PA_MODULE_AUTHOR("Colin Guthrie");
70 PA_MODULE_DESCRIPTION("RAOP Sink (Apple Airtunes)");
71 PA_MODULE_VERSION(PACKAGE_VERSION);
72 PA_MODULE_LOAD_ONCE(FALSE);
73 PA_MODULE_USAGE(
74         "sink_name=<name for the sink> "
75         "server=<address>  "
76         "format=<sample format> "
77         "channels=<number of channels> "
78         "rate=<sample rate>");
79
80 #define DEFAULT_SINK_NAME "airtunes"
81
82 struct userdata {
83     pa_core *core;
84     pa_module *module;
85     pa_sink *sink;
86
87     pa_thread_mq thread_mq;
88     pa_rtpoll *rtpoll;
89     pa_rtpoll_item *rtpoll_item;
90     pa_thread *thread;
91
92     pa_memchunk raw_memchunk;
93     pa_memchunk encoded_memchunk;
94
95     void *write_data;
96     size_t write_length, write_index;
97
98     void *read_data;
99     size_t read_length, read_index;
100
101     pa_usec_t latency;
102
103     /*esd_format_t format;*/
104     int32_t rate;
105
106     pa_smoother *smoother;
107     int fd;
108
109     int64_t offset;
110     int64_t encoding_overhead;
111     int32_t next_encoding_overhead;
112     double encoding_ratio;
113
114     pa_raop_client *raop;
115
116     size_t block_size;
117 };
118
119 static const char* const valid_modargs[] = {
120     "server",
121     "rate",
122     "format",
123     "channels",
124     "sink_name",
125     NULL
126 };
127
128 enum {
129     SINK_MESSAGE_PASS_SOCKET = PA_SINK_MESSAGE_MAX,
130     SINK_MESSAGE_RIP_SOCKET
131 };
132
133 /* Forward declaration */
134 static void sink_set_volume_cb(pa_sink *);
135
136 static void on_connection(PA_GCC_UNUSED int fd, void*userdata) {
137     struct userdata *u = userdata;
138     pa_assert(u);
139
140     pa_assert(u->fd < 0);
141     u->fd = fd;
142
143     /* Set the initial volume */
144     sink_set_volume_cb(u->sink);
145
146     pa_log_debug("Connection authenticated, handing fd to IO thread...");
147
148     pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_PASS_SOCKET, NULL, 0, NULL, NULL);
149 }
150
151 static void on_close(void*userdata) {
152     struct userdata *u = userdata;
153     pa_assert(u);
154
155     pa_log_debug("Connection closed, informing IO thread...");
156
157     pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_RIP_SOCKET, NULL, 0, NULL, NULL);
158 }
159
160 static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
161     struct userdata *u = PA_SINK(o)->userdata;
162
163     switch (code) {
164
165         case PA_SINK_MESSAGE_SET_STATE:
166
167             switch ((pa_sink_state_t) PA_PTR_TO_UINT(data)) {
168
169                 case PA_SINK_SUSPENDED:
170                     pa_assert(PA_SINK_IS_OPENED(u->sink->thread_info.state));
171
172                     pa_smoother_pause(u->smoother, pa_rtclock_usec());
173
174                     /* Issue a FLUSH if we are connected */
175                     if (u->fd >= 0) {
176                         pa_raop_flush(u->raop);
177                     }
178                     break;
179
180                 case PA_SINK_IDLE:
181                 case PA_SINK_RUNNING:
182
183                     if (u->sink->thread_info.state == PA_SINK_SUSPENDED) {
184                         pa_smoother_resume(u->smoother, pa_rtclock_usec());
185
186                         /* The connection can be closed when idle, so check to
187                            see if we need to reestablish it */
188                         if (u->fd < 0)
189                             pa_raop_connect(u->raop);
190                         else
191                             pa_raop_flush(u->raop);
192                     }
193
194                     break;
195
196                 case PA_SINK_UNLINKED:
197                 case PA_SINK_INIT:
198                 case PA_SINK_INVALID_STATE:
199                     ;
200             }
201
202             break;
203
204         case PA_SINK_MESSAGE_GET_LATENCY: {
205             pa_usec_t w, r;
206
207             r = pa_smoother_get(u->smoother, pa_rtclock_usec());
208             w = pa_bytes_to_usec((u->offset - u->encoding_overhead + (u->encoded_memchunk.length / u->encoding_ratio)), &u->sink->sample_spec);
209
210             *((pa_usec_t*) data) = w > r ? w - r : 0;
211             return 0;
212         }
213
214         case SINK_MESSAGE_PASS_SOCKET: {
215             struct pollfd *pollfd;
216
217             pa_assert(!u->rtpoll_item);
218
219             u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
220             pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
221             pollfd->fd = u->fd;
222             pollfd->events = POLLOUT;
223             /*pollfd->events = */pollfd->revents = 0;
224
225             if (u->sink->thread_info.state == PA_SINK_SUSPENDED) {
226                 /* Our stream has been suspended so we just flush it.... */
227                 pa_raop_flush(u->raop);
228             }
229             return 0;
230         }
231
232         case SINK_MESSAGE_RIP_SOCKET: {
233             pa_assert(u->fd >= 0);
234
235             pa_close(u->fd);
236             u->fd = -1;
237
238             if (u->sink->thread_info.state == PA_SINK_SUSPENDED) {
239
240                 pa_log_debug("RTSP control connection closed, but we're suspended so let's not worry about it... we'll open it again later");
241
242                 if (u->rtpoll_item)
243                     pa_rtpoll_item_free(u->rtpoll_item);
244                 u->rtpoll_item = NULL;
245             } else {
246                 /* Quesiton: is this valid here: or should we do some sort of:
247                    return pa_sink_process_msg(PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL);
248                    ?? */
249                 pa_module_unload_request(u->module, TRUE);
250             }
251             return 0;
252         }
253     }
254
255     return pa_sink_process_msg(o, code, data, offset, chunk);
256 }
257
258 static void sink_set_volume_cb(pa_sink *s) {
259     struct userdata *u = s->userdata;
260     pa_cvolume hw;
261     pa_volume_t v;
262     char t[PA_CVOLUME_SNPRINT_MAX];
263
264     pa_assert(u);
265
266     /* If we're muted we don't need to do anything */
267     if (s->muted)
268         return;
269
270     /* Calculate the max volume of all channels.
271        We'll use this as our (single) volume on the APEX device and emulate
272        any variation in channel volumes in software */
273     v = pa_cvolume_max(&s->virtual_volume);
274
275     /* Create a pa_cvolume version of our single value */
276     pa_cvolume_set(&hw, s->sample_spec.channels, v);
277
278     /* Perform any software manipulation of the volume needed */
279     pa_sw_cvolume_divide(&s->soft_volume, &s->virtual_volume, &hw);
280
281     pa_log_debug("Requested volume: %s", pa_cvolume_snprint(t, sizeof(t), &s->virtual_volume));
282     pa_log_debug("Got hardware volume: %s", pa_cvolume_snprint(t, sizeof(t), &hw));
283     pa_log_debug("Calculated software volume: %s", pa_cvolume_snprint(t, sizeof(t), &s->soft_volume));
284
285     /* Any necessary software volume manipulateion is done so set
286        our hw volume (or v as a single value) on the device */
287     pa_raop_client_set_volume(u->raop, v);
288 }
289
290 static void sink_set_mute_cb(pa_sink *s) {
291     struct userdata *u = s->userdata;
292
293     pa_assert(u);
294
295     if (s->muted) {
296         pa_raop_client_set_volume(u->raop, PA_VOLUME_MUTED);
297     } else {
298         sink_set_volume_cb(s);
299     }
300 }
301
302 static void thread_func(void *userdata) {
303     struct userdata *u = userdata;
304     int write_type = 0;
305     pa_memchunk silence;
306     uint32_t silence_overhead = 0;
307     double silence_ratio = 0;
308
309     pa_assert(u);
310
311     pa_log_debug("Thread starting up");
312
313     pa_thread_mq_install(&u->thread_mq);
314     pa_rtpoll_install(u->rtpoll);
315
316     pa_smoother_set_time_offset(u->smoother, pa_rtclock_usec());
317
318     /* Create a chunk of memory that is our encoded silence sample. */
319     pa_memchunk_reset(&silence);
320
321     for (;;) {
322         int ret;
323
324         if (PA_SINK_IS_OPENED(u->sink->thread_info.state))
325             if (u->sink->thread_info.rewind_requested)
326                 pa_sink_process_rewind(u->sink, 0);
327
328         if (u->rtpoll_item) {
329             struct pollfd *pollfd;
330             pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
331
332             /* Render some data and write it to the fifo */
333             if (/*PA_SINK_IS_OPENED(u->sink->thread_info.state) && */pollfd->revents) {
334                 pa_usec_t usec;
335                 int64_t n;
336                 void *p;
337
338                 if (!silence.memblock) {
339                     pa_memchunk silence_tmp;
340
341                     pa_memchunk_reset(&silence_tmp);
342                     silence_tmp.memblock = pa_memblock_new(u->core->mempool, 4096);
343                     silence_tmp.length = 4096;
344                     p = pa_memblock_acquire(silence_tmp.memblock);
345                       memset(p, 0, 4096);
346                     pa_memblock_release(silence_tmp.memblock);
347                     pa_raop_client_encode_sample(u->raop, &silence_tmp, &silence);
348                     pa_assert(0 == silence_tmp.length);
349                     silence_overhead = silence_tmp.length - 4096;
350                     silence_ratio = silence_tmp.length / 4096;
351                     pa_memblock_unref(silence_tmp.memblock);
352                 }
353
354                 for (;;) {
355                     ssize_t l;
356
357                     if (u->encoded_memchunk.length <= 0) {
358                         if (u->encoded_memchunk.memblock)
359                             pa_memblock_unref(u->encoded_memchunk.memblock);
360                         if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
361                             size_t rl;
362
363                             /* We render real data */
364                             if (u->raw_memchunk.length <= 0) {
365                                 if (u->raw_memchunk.memblock)
366                                     pa_memblock_unref(u->raw_memchunk.memblock);
367                                 pa_memchunk_reset(&u->raw_memchunk);
368
369                                 /* Grab unencoded data */
370                                 pa_sink_render(u->sink, u->block_size, &u->raw_memchunk);
371                             }
372                             pa_assert(u->raw_memchunk.length > 0);
373
374                             /* Encode it */
375                             rl = u->raw_memchunk.length;
376                             u->encoding_overhead += u->next_encoding_overhead;
377                             pa_raop_client_encode_sample(u->raop, &u->raw_memchunk, &u->encoded_memchunk);
378                             u->next_encoding_overhead = (u->encoded_memchunk.length - (rl - u->raw_memchunk.length));
379                             u->encoding_ratio = u->encoded_memchunk.length / (rl - u->raw_memchunk.length);
380                         } else {
381                             /* We render some silence into our memchunk */
382                             memcpy(&u->encoded_memchunk, &silence, sizeof(pa_memchunk));
383                             pa_memblock_ref(silence.memblock);
384
385                             /* Calculate/store some values to be used with the smoother */
386                             u->next_encoding_overhead = silence_overhead;
387                             u->encoding_ratio = silence_ratio;
388                         }
389                     }
390                     pa_assert(u->encoded_memchunk.length > 0);
391
392                     p = pa_memblock_acquire(u->encoded_memchunk.memblock);
393                     l = pa_write(u->fd, (uint8_t*) p + u->encoded_memchunk.index, u->encoded_memchunk.length, &write_type);
394                     pa_memblock_release(u->encoded_memchunk.memblock);
395
396                     pa_assert(l != 0);
397
398                     if (l < 0) {
399
400                         if (errno == EINTR)
401                             continue;
402                         else if (errno == EAGAIN) {
403
404                             /* OK, we filled all socket buffers up
405                              * now. */
406                             goto filled_up;
407
408                         } else {
409                             pa_log("Failed to write data to FIFO: %s", pa_cstrerror(errno));
410                             goto fail;
411                         }
412
413                     } else {
414                         u->offset += l;
415
416                         u->encoded_memchunk.index += l;
417                         u->encoded_memchunk.length -= l;
418
419                         pollfd->revents = 0;
420
421                         if (u->encoded_memchunk.length > 0) {
422                             /* we've completely written the encoded data, so update our overhead */
423                             u->encoding_overhead += u->next_encoding_overhead;
424
425                             /* OK, we wrote less that we asked for,
426                              * hence we can assume that the socket
427                              * buffers are full now */
428                             goto filled_up;
429                         }
430                     }
431                 }
432
433             filled_up:
434
435                 /* At this spot we know that the socket buffers are
436                  * fully filled up. This is the best time to estimate
437                  * the playback position of the server */
438
439                 n = u->offset - u->encoding_overhead;
440
441 #ifdef SIOCOUTQ
442                 {
443                     int l;
444                     if (ioctl(u->fd, SIOCOUTQ, &l) >= 0 && l > 0)
445                         n -= (l / u->encoding_ratio);
446                 }
447 #endif
448
449                 usec = pa_bytes_to_usec(n, &u->sink->sample_spec);
450
451                 if (usec > u->latency)
452                     usec -= u->latency;
453                 else
454                     usec = 0;
455
456                 pa_smoother_put(u->smoother, pa_rtclock_usec(), usec);
457             }
458
459             /* Hmm, nothing to do. Let's sleep */
460             pollfd->events = POLLOUT; /*PA_SINK_IS_OPENED(u->sink->thread_info.state)  ? POLLOUT : 0;*/
461         }
462
463         if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
464             goto fail;
465
466         if (ret == 0)
467             goto finish;
468
469         if (u->rtpoll_item) {
470             struct pollfd* pollfd;
471
472             pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
473
474             if (pollfd->revents & ~POLLOUT) {
475                 if (u->sink->thread_info.state != PA_SINK_SUSPENDED) {
476                     pa_log("FIFO shutdown.");
477                     goto fail;
478                 }
479
480                 /* We expect this to happen on occasion if we are not sending data.
481                    It's perfectly natural and normal and natural */
482                 if (u->rtpoll_item)
483                     pa_rtpoll_item_free(u->rtpoll_item);
484                 u->rtpoll_item = NULL;
485             }
486         }
487     }
488
489 fail:
490     /* If this was no regular exit from the loop we have to continue
491      * processing messages until we received PA_MESSAGE_SHUTDOWN */
492     pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
493     pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
494
495 finish:
496     if (silence.memblock)
497         pa_memblock_unref(silence.memblock);
498     pa_log_debug("Thread shutting down");
499 }
500
501 int pa__init(pa_module*m) {
502     struct userdata *u = NULL;
503     pa_sample_spec ss;
504     pa_modargs *ma = NULL;
505     const char *server;
506     pa_sink_new_data data;
507
508     pa_assert(m);
509
510     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
511         pa_log("failed to parse module arguments");
512         goto fail;
513     }
514
515     ss = m->core->default_sample_spec;
516     if (pa_modargs_get_sample_spec(ma, &ss) < 0) {
517         pa_log("invalid sample format specification");
518         goto fail;
519     }
520
521     if ((/*ss.format != PA_SAMPLE_U8 &&*/ ss.format != PA_SAMPLE_S16NE) ||
522         (ss.channels > 2)) {
523         pa_log("sample type support is limited to mono/stereo and U8 or S16NE sample data");
524         goto fail;
525     }
526
527     u = pa_xnew0(struct userdata, 1);
528     u->core = m->core;
529     u->module = m;
530     m->userdata = u;
531     u->fd = -1;
532     u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE, 10);
533     pa_memchunk_reset(&u->raw_memchunk);
534     pa_memchunk_reset(&u->encoded_memchunk);
535     u->offset = 0;
536     u->encoding_overhead = 0;
537     u->next_encoding_overhead = 0;
538     u->encoding_ratio = 1.0;
539
540     u->rtpoll = pa_rtpoll_new();
541     pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
542     u->rtpoll_item = NULL;
543
544     /*u->format =
545         (ss.format == PA_SAMPLE_U8 ? ESD_BITS8 : ESD_BITS16) |
546         (ss.channels == 2 ? ESD_STEREO : ESD_MONO);*/
547     u->rate = ss.rate;
548     u->block_size = pa_usec_to_bytes(PA_USEC_PER_SEC/20, &ss);
549
550     u->read_data = u->write_data = NULL;
551     u->read_index = u->write_index = u->read_length = u->write_length = 0;
552
553     /*u->state = STATE_AUTH;*/
554     u->latency = 0;
555
556     if (!(server = pa_modargs_get_value(ma, "server", NULL))) {
557         pa_log("No server argument given.");
558         goto fail;
559     }
560
561     pa_sink_new_data_init(&data);
562     data.driver = __FILE__;
563     data.module = m;
564     pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
565     pa_sink_new_data_set_sample_spec(&data, &ss);
566     pa_proplist_sets(data.proplist, PA_PROP_DEVICE_STRING, server);
567     pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "Airtunes sink '%s'", server);
568
569     u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY|PA_SINK_NETWORK);
570     pa_sink_new_data_done(&data);
571
572     if (!u->sink) {
573         pa_log("Failed to create sink.");
574         goto fail;
575     }
576
577     u->sink->parent.process_msg = sink_process_msg;
578     u->sink->userdata = u;
579     u->sink->set_volume = sink_set_volume_cb;
580     u->sink->set_mute = sink_set_mute_cb;
581     u->sink->flags = PA_SINK_LATENCY|PA_SINK_NETWORK|PA_SINK_HW_VOLUME_CTRL;
582
583     pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
584     pa_sink_set_rtpoll(u->sink, u->rtpoll);
585
586     if (!(u->raop = pa_raop_client_new(u->core, server))) {
587         pa_log("Failed to connect to server.");
588         goto fail;
589     }
590
591     pa_raop_client_set_callback(u->raop, on_connection, u);
592     pa_raop_client_set_closed_callback(u->raop, on_close, u);
593
594     if (!(u->thread = pa_thread_new(thread_func, u))) {
595         pa_log("Failed to create thread.");
596         goto fail;
597     }
598
599     pa_sink_put(u->sink);
600
601     pa_modargs_free(ma);
602
603     return 0;
604
605 fail:
606     if (ma)
607         pa_modargs_free(ma);
608
609     pa__done(m);
610
611     return -1;
612 }
613
614 int pa__get_n_used(pa_module *m) {
615     struct userdata *u;
616
617     pa_assert(m);
618     pa_assert_se(u = m->userdata);
619
620     return pa_sink_linked_by(u->sink);
621 }
622
623 void pa__done(pa_module*m) {
624     struct userdata *u;
625     pa_assert(m);
626
627     if (!(u = m->userdata))
628         return;
629
630     if (u->sink)
631         pa_sink_unlink(u->sink);
632
633     if (u->thread) {
634         pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
635         pa_thread_free(u->thread);
636     }
637
638     pa_thread_mq_done(&u->thread_mq);
639
640     if (u->sink)
641         pa_sink_unref(u->sink);
642
643     if (u->rtpoll_item)
644         pa_rtpoll_item_free(u->rtpoll_item);
645
646     if (u->rtpoll)
647         pa_rtpoll_free(u->rtpoll);
648
649     if (u->raw_memchunk.memblock)
650         pa_memblock_unref(u->raw_memchunk.memblock);
651
652     if (u->encoded_memchunk.memblock)
653         pa_memblock_unref(u->encoded_memchunk.memblock);
654
655     if (u->raop)
656         pa_raop_client_free(u->raop);
657
658     pa_xfree(u->read_data);
659     pa_xfree(u->write_data);
660
661     if (u->smoother)
662         pa_smoother_free(u->smoother);
663
664     if (u->fd >= 0)
665         pa_close(u->fd);
666
667     pa_xfree(u);
668 }