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/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->virtual_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->virtual_volume, &hw);
 293
 294     pa_log_debug("Requested volume: %s", pa_cvolume_snprint(t, sizeof(t), &s->virtual_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     pa_rtpoll_install(u->rtpoll);
 328
 329     pa_smoother_set_time_offset(u->smoother, pa_rtclock_now());
 330
 331     /* Create a chunk of memory that is our encoded silence sample. */
 332     pa_memchunk_reset(&silence);
 333
 334     for (;;) {
 335         int ret;
 336
 337         if (PA_SINK_IS_OPENED(u->sink->thread_info.state))
 338             if (u->sink->thread_info.rewind_requested)
 339                 pa_sink_process_rewind(u->sink, 0);
 340
 341         if (u->rtpoll_item) {
 342             struct pollfd *pollfd;
 343             pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
 344
 345             /* Render some data and write it to the fifo */
 346             if (/*PA_SINK_IS_OPENED(u->sink->thread_info.state) && */pollfd->revents) {
 347                 pa_usec_t usec;
 348                 int64_t n;
 349                 void *p;
 350
 351                 if (!silence.memblock) {
 352                     pa_memchunk silence_tmp;
 353
 354                     pa_memchunk_reset(&silence_tmp);
 355                     silence_tmp.memblock = pa_memblock_new(u->core->mempool, 4096);
 356                     silence_tmp.length = 4096;
 357                     p = pa_memblock_acquire(silence_tmp.memblock);
 358                       memset(p, 0, 4096);
 359                     pa_memblock_release(silence_tmp.memblock);
 360                     pa_raop_client_encode_sample(u->raop, &silence_tmp, &silence);
 361                     pa_assert(0 == silence_tmp.length);
 362                     silence_overhead = silence_tmp.length - 4096;
 363                     silence_ratio = silence_tmp.length / 4096;
 364                     pa_memblock_unref(silence_tmp.memblock);
 365                 }
 366
 367                 for (;;) {
 368                     ssize_t l;
 369
 370                     if (u->encoded_memchunk.length <= 0) {
 371                         if (u->encoded_memchunk.memblock)
 372                             pa_memblock_unref(u->encoded_memchunk.memblock);
 373                         if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
 374                             size_t rl;
 375
 376                             /* We render real data */
 377                             if (u->raw_memchunk.length <= 0) {
 378                                 if (u->raw_memchunk.memblock)
 379                                     pa_memblock_unref(u->raw_memchunk.memblock);
 380                                 pa_memchunk_reset(&u->raw_memchunk);
 381
 382                                 /* Grab unencoded data */
 383                                 pa_sink_render(u->sink, u->block_size, &u->raw_memchunk);
 384                             }
 385                             pa_assert(u->raw_memchunk.length > 0);
 386
 387                             /* Encode it */
 388                             rl = u->raw_memchunk.length;
 389                             u->encoding_overhead += u->next_encoding_overhead;
 390                             pa_raop_client_encode_sample(u->raop, &u->raw_memchunk, &u->encoded_memchunk);
 391                             u->next_encoding_overhead = (u->encoded_memchunk.length - (rl - u->raw_memchunk.length));
 392                             u->encoding_ratio = u->encoded_memchunk.length / (rl - u->raw_memchunk.length);
 393                         } else {
 394                             /* We render some silence into our memchunk */
 395                             memcpy(&u->encoded_memchunk, &silence, sizeof(pa_memchunk));
 396                             pa_memblock_ref(silence.memblock);
 397
 398                             /* Calculate/store some values to be used with the smoother */
 399                             u->next_encoding_overhead = silence_overhead;
 400                             u->encoding_ratio = silence_ratio;
 401                         }
 402                     }
 403                     pa_assert(u->encoded_memchunk.length > 0);
 404
 405                     p = pa_memblock_acquire(u->encoded_memchunk.memblock);
 406                     l = pa_write(u->fd, (uint8_t*) p + u->encoded_memchunk.index, u->encoded_memchunk.length, &write_type);
 407                     pa_memblock_release(u->encoded_memchunk.memblock);
 408
 409                     pa_assert(l != 0);
 410
 411                     if (l < 0) {
 412
 413                         if (errno == EINTR)
 414                             continue;
 415                         else if (errno == EAGAIN) {
 416
 417                             /* OK, we filled all socket buffers up
 418                              * now. */
 419                             goto filled_up;
 420
 421                         } else {
 422                             pa_log("Failed to write data to FIFO: %s", pa_cstrerror(errno));
 423                             goto fail;
 424                         }
 425
 426                     } else {
 427                         u->offset += l;
 428
 429                         u->encoded_memchunk.index += l;
 430                         u->encoded_memchunk.length -= l;
 431
 432                         pollfd->revents = 0;
 433
 434                         if (u->encoded_memchunk.length > 0) {
 435                             /* we've completely written the encoded data, so update our overhead */
 436                             u->encoding_overhead += u->next_encoding_overhead;
 437
 438                             /* OK, we wrote less that we asked for,
 439                              * hence we can assume that the socket
 440                              * buffers are full now */
 441                             goto filled_up;
 442                         }
 443                     }
 444                 }
 445
 446             filled_up:
 447
 448                 /* At this spot we know that the socket buffers are
 449                  * fully filled up. This is the best time to estimate
 450                  * the playback position of the server */
 451
 452                 n = u->offset - u->encoding_overhead;
 453
 454 #ifdef SIOCOUTQ
 455                 {
 456                     int l;
 457                     if (ioctl(u->fd, SIOCOUTQ, &l) >= 0 && l > 0)
 458                         n -= (l / u->encoding_ratio);
 459                 }
 460 #endif
 461
 462                 usec = pa_bytes_to_usec(n, &u->sink->sample_spec);
 463
 464                 if (usec > u->latency)
 465                     usec -= u->latency;
 466                 else
 467                     usec = 0;
 468
 469                 pa_smoother_put(u->smoother, pa_rtclock_now(), usec);
 470             }
 471
 472             /* Hmm, nothing to do. Let's sleep */
 473             pollfd->events = POLLOUT; /*PA_SINK_IS_OPENED(u->sink->thread_info.state)  ? POLLOUT : 0;*/
 474         }
 475
 476         if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
 477             goto fail;
 478
 479         if (ret == 0)
 480             goto finish;
 481
 482         if (u->rtpoll_item) {
 483             struct pollfd* pollfd;
 484
 485             pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
 486
 487             if (pollfd->revents & ~POLLOUT) {
 488                 if (u->sink->thread_info.state != PA_SINK_SUSPENDED) {
 489                     pa_log("FIFO shutdown.");
 490                     goto fail;
 491                 }
 492
 493                 /* We expect this to happen on occasion if we are not sending data.
 494                    It's perfectly natural and normal and natural */
 495                 if (u->rtpoll_item)
 496                     pa_rtpoll_item_free(u->rtpoll_item);
 497                 u->rtpoll_item = NULL;
 498             }
 499         }
 500     }
 501
 502 fail:
 503     /* If this was no regular exit from the loop we have to continue
 504      * processing messages until we received PA_MESSAGE_SHUTDOWN */
 505     pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
 506     pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
 507
 508 finish:
 509     if (silence.memblock)
 510         pa_memblock_unref(silence.memblock);
 511     pa_log_debug("Thread shutting down");
 512 }
 513
 514 int pa__init(pa_module*m) {
 515     struct userdata *u = NULL;
 516     pa_sample_spec ss;
 517     pa_modargs *ma = NULL;
 518     const char *server, *desc;
 519     pa_sink_new_data data;
 520
 521     pa_assert(m);
 522
 523     if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
 524         pa_log("failed to parse module arguments");
 525         goto fail;
 526     }
 527
 528     ss = m->core->default_sample_spec;
 529     if (pa_modargs_get_sample_spec(ma, &ss) < 0) {
 530         pa_log("invalid sample format specification");
 531         goto fail;
 532     }
 533
 534     if ((/*ss.format != PA_SAMPLE_U8 &&*/ ss.format != PA_SAMPLE_S16NE) ||
 535         (ss.channels > 2)) {
 536         pa_log("sample type support is limited to mono/stereo and U8 or S16NE sample data");
 537         goto fail;
 538     }
 539
 540     u = pa_xnew0(struct userdata, 1);
 541     u->core = m->core;
 542     u->module = m;
 543     m->userdata = u;
 544     u->fd = -1;
 545     u->smoother = pa_smoother_new(
 546             PA_USEC_PER_SEC,
 547             PA_USEC_PER_SEC*2,
 548             TRUE,
 549             TRUE,
 550             10,
 551             0,
 552             FALSE);
 553     pa_memchunk_reset(&u->raw_memchunk);
 554     pa_memchunk_reset(&u->encoded_memchunk);
 555     u->offset = 0;
 556     u->encoding_overhead = 0;
 557     u->next_encoding_overhead = 0;
 558     u->encoding_ratio = 1.0;
 559
 560     u->rtpoll = pa_rtpoll_new();
 561     pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
 562     u->rtpoll_item = NULL;
 563
 564     /*u->format =
 565         (ss.format == PA_SAMPLE_U8 ? ESD_BITS8 : ESD_BITS16) |
 566         (ss.channels == 2 ? ESD_STEREO : ESD_MONO);*/
 567     u->rate = ss.rate;
 568     u->block_size = pa_usec_to_bytes(PA_USEC_PER_SEC/20, &ss);
 569
 570     u->read_data = u->write_data = NULL;
 571     u->read_index = u->write_index = u->read_length = u->write_length = 0;
 572
 573     /*u->state = STATE_AUTH;*/
 574     u->latency = 0;
 575
 576     if (!(server = pa_modargs_get_value(ma, "server", NULL))) {
 577         pa_log("No server argument given.");
 578         goto fail;
 579     }
 580
 581     pa_sink_new_data_init(&data);
 582     data.driver = __FILE__;
 583     data.module = m;
 584     pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
 585     pa_sink_new_data_set_sample_spec(&data, &ss);
 586     pa_proplist_sets(data.proplist, PA_PROP_DEVICE_STRING, server);
 587     pa_proplist_sets(data.proplist, PA_PROP_DEVICE_INTENDED_ROLES, "music");
 588     if ((desc = pa_modargs_get_value(ma, "description", NULL)))
 589         pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, desc);
 590     else
 591         pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "RAOP sink '%s'", server);
 592
 593     if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
 594         pa_log("Invalid properties");
 595         pa_sink_new_data_done(&data);
 596         goto fail;
 597     }
 598
 599     u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY|PA_SINK_NETWORK);
 600     pa_sink_new_data_done(&data);
 601
 602     if (!u->sink) {
 603         pa_log("Failed to create sink.");
 604         goto fail;
 605     }
 606
 607     u->sink->parent.process_msg = sink_process_msg;
 608     u->sink->userdata = u;
 609     u->sink->set_volume = sink_set_volume_cb;
 610     u->sink->set_mute = sink_set_mute_cb;
 611     u->sink->flags = PA_SINK_LATENCY|PA_SINK_NETWORK|PA_SINK_HW_VOLUME_CTRL;
 612
 613     pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
 614     pa_sink_set_rtpoll(u->sink, u->rtpoll);
 615
 616     if (!(u->raop = pa_raop_client_new(u->core, server))) {
 617         pa_log("Failed to connect to server.");
 618         goto fail;
 619     }
 620
 621     pa_raop_client_set_callback(u->raop, on_connection, u);
 622     pa_raop_client_set_closed_callback(u->raop, on_close, u);
 623
 624     if (!(u->thread = pa_thread_new(thread_func, u))) {
 625         pa_log("Failed to create thread.");
 626         goto fail;
 627     }
 628
 629     pa_sink_put(u->sink);
 630
 631     pa_modargs_free(ma);
 632
 633     return 0;
 634
 635 fail:
 636     if (ma)
 637         pa_modargs_free(ma);
 638
 639     pa__done(m);
 640
 641     return -1;
 642 }
 643
 644 int pa__get_n_used(pa_module *m) {
 645     struct userdata *u;
 646
 647     pa_assert(m);
 648     pa_assert_se(u = m->userdata);
 649
 650     return pa_sink_linked_by(u->sink);
 651 }
 652
 653 void pa__done(pa_module*m) {
 654     struct userdata *u;
 655     pa_assert(m);
 656
 657     if (!(u = m->userdata))
 658         return;
 659
 660     if (u->sink)
 661         pa_sink_unlink(u->sink);
 662
 663     if (u->thread) {
 664         pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
 665         pa_thread_free(u->thread);
 666     }
 667
 668     pa_thread_mq_done(&u->thread_mq);
 669
 670     if (u->sink)
 671         pa_sink_unref(u->sink);
 672
 673     if (u->rtpoll_item)
 674         pa_rtpoll_item_free(u->rtpoll_item);
 675
 676     if (u->rtpoll)
 677         pa_rtpoll_free(u->rtpoll);
 678
 679     if (u->raw_memchunk.memblock)
 680         pa_memblock_unref(u->raw_memchunk.memblock);
 681
 682     if (u->encoded_memchunk.memblock)
 683         pa_memblock_unref(u->encoded_memchunk.memblock);
 684
 685     if (u->raop)
 686         pa_raop_client_free(u->raop);
 687
 688     pa_xfree(u->read_data);
 689     pa_xfree(u->write_data);
 690
 691     if (u->smoother)
 692         pa_smoother_free(u->smoother);
 693
 694     if (u->fd >= 0)
 695         pa_close(u->fd);
 696
 697     pa_xfree(u);
 698 }