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