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