]> code.delx.au - pulseaudio/blob - src/modules/module-tunnel.c
delx.au / code / pulseaudio / blob
summary | shortlog | log | commit | commitdiff | tree
history | raw | HEAD
sink: volume handling rework, new flat volume logic
[pulseaudio] / src / modules / module-tunnel.c
1 /***
2 This file is part of PulseAudio.
3
4 Copyright 2004-2006 Lennart Poettering
5 Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
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 <unistd.h>
28 #include <string.h>
29 #include <errno.h>
30 #include <sys/types.h>
31 #include <stdio.h>
32 #include <stdlib.h>
33
34 #include <pulse/rtclock.h>
35 #include <pulse/timeval.h>
36 #include <pulse/util.h>
37 #include <pulse/version.h>
38 #include <pulse/xmalloc.h>
39
40 #include <pulsecore/module.h>
41 #include <pulsecore/core-util.h>
42 #include <pulsecore/modargs.h>
43 #include <pulsecore/log.h>
44 #include <pulsecore/core-subscribe.h>
45 #include <pulsecore/sink-input.h>
46 #include <pulsecore/pdispatch.h>
47 #include <pulsecore/pstream.h>
48 #include <pulsecore/pstream-util.h>
49 #include <pulsecore/socket-client.h>
50 #include <pulsecore/socket-util.h>
51 #include <pulsecore/time-smoother.h>
52 #include <pulsecore/thread.h>
53 #include <pulsecore/thread-mq.h>
54 #include <pulsecore/core-rtclock.h>
55 #include <pulsecore/core-error.h>
56 #include <pulsecore/proplist-util.h>
57 #include <pulsecore/auth-cookie.h>
58 #include <pulsecore/mcalign.h>
59
60 #ifdef TUNNEL_SINK
61 #include "module-tunnel-sink-symdef.h"
62 #else
63 #include "module-tunnel-source-symdef.h"
64 #endif
65
66 #ifdef TUNNEL_SINK
67 PA_MODULE_DESCRIPTION("Tunnel module for sinks");
68 PA_MODULE_USAGE(
69 "sink_name=<name for the local sink> "
70 "sink_properties=<properties for the local sink> "
71 "server=<address> "
72 "sink=<remote sink name> "
73 "cookie=<filename> "
74 "format=<sample format> "
75 "channels=<number of channels> "
76 "rate=<sample rate> "
77 "channel_map=<channel map>");
78 #else
79 PA_MODULE_DESCRIPTION("Tunnel module for sources");
80 PA_MODULE_USAGE(
81 "source_name=<name for the local source> "
82 "source_properties=<properties for the local source> "
83 "server=<address> "
84 "source=<remote source name> "
85 "cookie=<filename> "
86 "format=<sample format> "
87 "channels=<number of channels> "
88 "rate=<sample rate> "
89 "channel_map=<channel map>");
90 #endif
91
92 PA_MODULE_AUTHOR("Lennart Poettering");
93 PA_MODULE_VERSION(PACKAGE_VERSION);
94 PA_MODULE_LOAD_ONCE(FALSE);
95
96 static const char* const valid_modargs[] = {
97 "server",
98 "cookie",
99 "format",
100 "channels",
101 "rate",
102 #ifdef TUNNEL_SINK
103 "sink_name",
104 "sink_properties",
105 "sink",
106 #else
107 "source_name",
108 "source_properties",
109 "source",
110 #endif
111 "channel_map",
112 NULL,
113 };
114
115 #define DEFAULT_TIMEOUT 5
116
117 #define LATENCY_INTERVAL (10*PA_USEC_PER_SEC)
118
119 #define MIN_NETWORK_LATENCY_USEC (8*PA_USEC_PER_MSEC)
120
121 #ifdef TUNNEL_SINK
122
123 enum {
124 SINK_MESSAGE_REQUEST = PA_SINK_MESSAGE_MAX,
125 SINK_MESSAGE_REMOTE_SUSPEND,
126 SINK_MESSAGE_UPDATE_LATENCY,
127 SINK_MESSAGE_POST
128 };
129
130 #define DEFAULT_TLENGTH_MSEC 150
131 #define DEFAULT_MINREQ_MSEC 25
132
133 #else
134
135 enum {
136 SOURCE_MESSAGE_POST = PA_SOURCE_MESSAGE_MAX,
137 SOURCE_MESSAGE_REMOTE_SUSPEND,
138 SOURCE_MESSAGE_UPDATE_LATENCY
139 };
140
141 #define DEFAULT_FRAGSIZE_MSEC 25
142
143 #endif
144
145 #ifdef TUNNEL_SINK
146 static void command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
147 static void command_started(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
148 #endif
149 static void command_subscribe_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
150 static void command_stream_killed(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
151 static void command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
152 static void command_suspended(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
153 static void command_moved(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
154 static void command_stream_or_client_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
155 static void command_stream_buffer_attr_changed(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
156
157 static const pa_pdispatch_cb_t command_table[PA_COMMAND_MAX] = {
158 #ifdef TUNNEL_SINK
159 [PA_COMMAND_REQUEST] = command_request,
160 [PA_COMMAND_STARTED] = command_started,
161 #endif
162 [PA_COMMAND_SUBSCRIBE_EVENT] = command_subscribe_event,
163 [PA_COMMAND_OVERFLOW] = command_overflow_or_underflow,
164 [PA_COMMAND_UNDERFLOW] = command_overflow_or_underflow,
165 [PA_COMMAND_PLAYBACK_STREAM_KILLED] = command_stream_killed,
166 [PA_COMMAND_RECORD_STREAM_KILLED] = command_stream_killed,
167 [PA_COMMAND_PLAYBACK_STREAM_SUSPENDED] = command_suspended,
168 [PA_COMMAND_RECORD_STREAM_SUSPENDED] = command_suspended,
169 [PA_COMMAND_PLAYBACK_STREAM_MOVED] = command_moved,
170 [PA_COMMAND_RECORD_STREAM_MOVED] = command_moved,
171 [PA_COMMAND_PLAYBACK_STREAM_EVENT] = command_stream_or_client_event,
172 [PA_COMMAND_RECORD_STREAM_EVENT] = command_stream_or_client_event,
173 [PA_COMMAND_CLIENT_EVENT] = command_stream_or_client_event,
174 [PA_COMMAND_PLAYBACK_BUFFER_ATTR_CHANGED] = command_stream_buffer_attr_changed,
175 [PA_COMMAND_RECORD_BUFFER_ATTR_CHANGED] = command_stream_buffer_attr_changed
176 };
177
178 struct userdata {
179 pa_core *core;
180 pa_module *module;
181
182 pa_thread_mq thread_mq;
183 pa_rtpoll *rtpoll;
184 pa_thread *thread;
185
186 pa_socket_client *client;
187 pa_pstream *pstream;
188 pa_pdispatch *pdispatch;
189
190 char *server_name;
191 #ifdef TUNNEL_SINK
192 char *sink_name;
193 pa_sink *sink;
194 size_t requested_bytes;
195 #else
196 char *source_name;
197 pa_source *source;
198 pa_mcalign *mcalign;
199 #endif
200
201 pa_auth_cookie *auth_cookie;
202
203 uint32_t version;
204 uint32_t ctag;
205 uint32_t device_index;
206 uint32_t channel;
207
208 int64_t counter, counter_delta;
209
210 pa_bool_t remote_corked:1;
211 pa_bool_t remote_suspended:1;
212
213 pa_usec_t transport_usec; /* maintained in the main thread */
214 pa_usec_t thread_transport_usec; /* maintained in the IO thread */
215
216 uint32_t ignore_latency_before;
217
218 pa_time_event *time_event;
219
220 pa_smoother *smoother;
221
222 char *device_description;
223 char *server_fqdn;
224 char *user_name;
225
226 uint32_t maxlength;
227 #ifdef TUNNEL_SINK
228 uint32_t tlength;
229 uint32_t minreq;
230 uint32_t prebuf;
231 #else
232 uint32_t fragsize;
233 #endif
234 };
235
236 static void request_latency(struct userdata *u);
237
238 /* Called from main context */
239 static void command_stream_or_client_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
240 pa_log_debug("Got stream or client event.");
241 }
242
243 /* Called from main context */
244 static void command_stream_killed(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
245 struct userdata *u = userdata;
246
247 pa_assert(pd);
248 pa_assert(t);
249 pa_assert(u);
250 pa_assert(u->pdispatch == pd);
251
252 pa_log_warn("Stream killed");
253 pa_module_unload_request(u->module, TRUE);
254 }
255
256 /* Called from main context */
257 static void command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
258 struct userdata *u = userdata;
259
260 pa_assert(pd);
261 pa_assert(t);
262 pa_assert(u);
263 pa_assert(u->pdispatch == pd);
264
265 pa_log_info("Server signalled buffer overrun/underrun.");
266 request_latency(u);
267 }
268
269 /* Called from main context */
270 static void command_suspended(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
271 struct userdata *u = userdata;
272 uint32_t channel;
273 pa_bool_t suspended;
274
275 pa_assert(pd);
276 pa_assert(t);
277 pa_assert(u);
278 pa_assert(u->pdispatch == pd);
279
280 if (pa_tagstruct_getu32(t, &channel) < 0 ||
281 pa_tagstruct_get_boolean(t, &suspended) < 0 ||
282 !pa_tagstruct_eof(t)) {
283
284 pa_log("Invalid packet.");
285 pa_module_unload_request(u->module, TRUE);
286 return;
287 }
288
289 pa_log_debug("Server reports device suspend.");
290
291 #ifdef TUNNEL_SINK
292 pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REMOTE_SUSPEND, PA_UINT32_TO_PTR(!!suspended), 0, NULL);
293 #else
294 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_REMOTE_SUSPEND, PA_UINT32_TO_PTR(!!suspended), 0, NULL);
295 #endif
296
297 request_latency(u);
298 }
299
300 /* Called from main context */
301 static void command_moved(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
302 struct userdata *u = userdata;
303 uint32_t channel, di;
304 const char *dn;
305 pa_bool_t suspended;
306
307 pa_assert(pd);
308 pa_assert(t);
309 pa_assert(u);
310 pa_assert(u->pdispatch == pd);
311
312 if (pa_tagstruct_getu32(t, &channel) < 0 ||
313 pa_tagstruct_getu32(t, &di) < 0 ||
314 pa_tagstruct_gets(t, &dn) < 0 ||
315 pa_tagstruct_get_boolean(t, &suspended) < 0) {
316
317 pa_log_error("Invalid packet.");
318 pa_module_unload_request(u->module, TRUE);
319 return;
320 }
321
322 pa_log_debug("Server reports a stream move.");
323
324 #ifdef TUNNEL_SINK
325 pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REMOTE_SUSPEND, PA_UINT32_TO_PTR(!!suspended), 0, NULL);
326 #else
327 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_REMOTE_SUSPEND, PA_UINT32_TO_PTR(!!suspended), 0, NULL);
328 #endif
329
330 request_latency(u);
331 }
332
333 static void command_stream_buffer_attr_changed(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
334 struct userdata *u = userdata;
335 uint32_t channel, maxlength, tlength, fragsize, prebuf, minreq;
336 pa_usec_t usec;
337
338 pa_assert(pd);
339 pa_assert(t);
340 pa_assert(u);
341 pa_assert(u->pdispatch == pd);
342
343 if (pa_tagstruct_getu32(t, &channel) < 0 ||
344 pa_tagstruct_getu32(t, &maxlength) < 0) {
345
346 pa_log_error("Invalid packet.");
347 pa_module_unload_request(u->module, TRUE);
348 return;
349 }
350
351 if (command == PA_COMMAND_RECORD_BUFFER_ATTR_CHANGED) {
352 if (pa_tagstruct_getu32(t, &fragsize) < 0 ||
353 pa_tagstruct_get_usec(t, &usec) < 0) {
354
355 pa_log_error("Invalid packet.");
356 pa_module_unload_request(u->module, TRUE);
357 return;
358 }
359 } else {
360 if (pa_tagstruct_getu32(t, &tlength) < 0 ||
361 pa_tagstruct_getu32(t, &prebuf) < 0 ||
362 pa_tagstruct_getu32(t, &minreq) < 0 ||
363 pa_tagstruct_get_usec(t, &usec) < 0) {
364
365 pa_log_error("Invalid packet.");
366 pa_module_unload_request(u->module, TRUE);
367 return;
368 }
369 }
370
371 #ifdef TUNNEL_SINK
372 pa_log_debug("Server reports buffer attrs changed. tlength now at %lu, before %lu.", (unsigned long) tlength, (unsigned long) u->tlength);
373 #endif
374
375 request_latency(u);
376 }
377
378 #ifdef TUNNEL_SINK
379
380 /* Called from main context */
381 static void command_started(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
382 struct userdata *u = userdata;
383
384 pa_assert(pd);
385 pa_assert(t);
386 pa_assert(u);
387 pa_assert(u->pdispatch == pd);
388
389 pa_log_debug("Server reports playback started.");
390 request_latency(u);
391 }
392
393 #endif
394
395 /* Called from IO thread context */
396 static void check_smoother_status(struct userdata *u, pa_bool_t past) {
397 pa_usec_t x;
398
399 pa_assert(u);
400
401 x = pa_rtclock_now();
402
403 /* Correct by the time the requested issued needs to travel to the
404 * other side. This is a valid thread-safe access, because the
405 * main thread is waiting for us */
406
407 if (past)
408 x -= u->thread_transport_usec;
409 else
410 x += u->thread_transport_usec;
411
412 if (u->remote_suspended || u->remote_corked)
413 pa_smoother_pause(u->smoother, x);
414 else
415 pa_smoother_resume(u->smoother, x, TRUE);
416 }
417
418 /* Called from IO thread context */
419 static void stream_cork_within_thread(struct userdata *u, pa_bool_t cork) {
420 pa_assert(u);
421
422 if (u->remote_corked == cork)
423 return;
424
425 u->remote_corked = cork;
426 check_smoother_status(u, FALSE);
427 }
428
429 /* Called from main context */
430 static void stream_cork(struct userdata *u, pa_bool_t cork) {
431 pa_tagstruct *t;
432 pa_assert(u);
433
434 if (!u->pstream)
435 return;
436
437 t = pa_tagstruct_new(NULL, 0);
438 #ifdef TUNNEL_SINK
439 pa_tagstruct_putu32(t, PA_COMMAND_CORK_PLAYBACK_STREAM);
440 #else
441 pa_tagstruct_putu32(t, PA_COMMAND_CORK_RECORD_STREAM);
442 #endif
443 pa_tagstruct_putu32(t, u->ctag++);
444 pa_tagstruct_putu32(t, u->channel);
445 pa_tagstruct_put_boolean(t, !!cork);
446 pa_pstream_send_tagstruct(u->pstream, t);
447
448 request_latency(u);
449 }
450
451 /* Called from IO thread context */
452 static void stream_suspend_within_thread(struct userdata *u, pa_bool_t suspend) {
453 pa_assert(u);
454
455 if (u->remote_suspended == suspend)
456 return;
457
458 u->remote_suspended = suspend;
459 check_smoother_status(u, TRUE);
460 }
461
462 #ifdef TUNNEL_SINK
463
464 /* Called from IO thread context */
465 static void send_data(struct userdata *u) {
466 pa_assert(u);
467
468 while (u->requested_bytes > 0) {
469 pa_memchunk memchunk;
470
471 pa_sink_render(u->sink, u->requested_bytes, &memchunk);
472 pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_POST, NULL, 0, &memchunk, NULL);
473 pa_memblock_unref(memchunk.memblock);
474
475 u->requested_bytes -= memchunk.length;
476
477 u->counter += (int64_t) memchunk.length;
478 }
479 }
480
481 /* This function is called from IO context -- except when it is not. */
482 static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
483 struct userdata *u = PA_SINK(o)->userdata;
484
485 switch (code) {
486
487 case PA_SINK_MESSAGE_SET_STATE: {
488 int r;
489
490 /* First, change the state, because otherwide pa_sink_render() would fail */
491 if ((r = pa_sink_process_msg(o, code, data, offset, chunk)) >= 0) {
492
493 stream_cork_within_thread(u, u->sink->state == PA_SINK_SUSPENDED);
494
495 if (PA_SINK_IS_OPENED(u->sink->state))
496 send_data(u);
497 }
498
499 return r;
500 }
501
502 case PA_SINK_MESSAGE_GET_LATENCY: {
503 pa_usec_t yl, yr, *usec = data;
504
505 yl = pa_bytes_to_usec((uint64_t) u->counter, &u->sink->sample_spec);
506 yr = pa_smoother_get(u->smoother, pa_rtclock_now());
507
508 *usec = yl > yr ? yl - yr : 0;
509 return 0;
510 }
511
512 case SINK_MESSAGE_REQUEST:
513
514 pa_assert(offset > 0);
515 u->requested_bytes += (size_t) offset;
516
517 if (PA_SINK_IS_OPENED(u->sink->thread_info.state))
518 send_data(u);
519
520 return 0;
521
522
523 case SINK_MESSAGE_REMOTE_SUSPEND:
524
525 stream_suspend_within_thread(u, !!PA_PTR_TO_UINT(data));
526 return 0;
527
528
529 case SINK_MESSAGE_UPDATE_LATENCY: {
530 pa_usec_t y;
531
532 y = pa_bytes_to_usec((uint64_t) u->counter, &u->sink->sample_spec);
533
534 if (y > (pa_usec_t) offset)
535 y -= (pa_usec_t) offset;
536 else
537 y = 0;
538
539 pa_smoother_put(u->smoother, pa_rtclock_now(), y);
540
541 /* We can access this freely here, since the main thread is waiting for us */
542 u->thread_transport_usec = u->transport_usec;
543
544 return 0;
545 }
546
547 case SINK_MESSAGE_POST:
548
549 /* OK, This might be a bit confusing. This message is
550 * delivered to us from the main context -- NOT from the
551 * IO thread context where the rest of the messages are
552 * dispatched. Yeah, ugly, but I am a lazy bastard. */
553
554 pa_pstream_send_memblock(u->pstream, u->channel, 0, PA_SEEK_RELATIVE, chunk);
555
556 u->counter_delta += (int64_t) chunk->length;
557
558 return 0;
559 }
560
561 return pa_sink_process_msg(o, code, data, offset, chunk);
562 }
563
564 /* Called from main context */
565 static int sink_set_state(pa_sink *s, pa_sink_state_t state) {
566 struct userdata *u;
567 pa_sink_assert_ref(s);
568 u = s->userdata;
569
570 switch ((pa_sink_state_t) state) {
571
572 case PA_SINK_SUSPENDED:
573 pa_assert(PA_SINK_IS_OPENED(s->state));
574 stream_cork(u, TRUE);
575 break;
576
577 case PA_SINK_IDLE:
578 case PA_SINK_RUNNING:
579 if (s->state == PA_SINK_SUSPENDED)
580 stream_cork(u, FALSE);
581 break;
582
583 case PA_SINK_UNLINKED:
584 case PA_SINK_INIT:
585 case PA_SINK_INVALID_STATE:
586 ;
587 }
588
589 return 0;
590 }
591
592 #else
593
594 /* This function is called from IO context -- except when it is not. */
595 static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
596 struct userdata *u = PA_SOURCE(o)->userdata;
597
598 switch (code) {
599
600 case PA_SOURCE_MESSAGE_SET_STATE: {
601 int r;
602
603 if ((r = pa_source_process_msg(o, code, data, offset, chunk)) >= 0)
604 stream_cork_within_thread(u, u->source->state == PA_SOURCE_SUSPENDED);
605
606 return r;
607 }
608
609 case PA_SOURCE_MESSAGE_GET_LATENCY: {
610 pa_usec_t yr, yl, *usec = data;
611
612 yl = pa_bytes_to_usec((uint64_t) u->counter, &PA_SOURCE(o)->sample_spec);
613 yr = pa_smoother_get(u->smoother, pa_rtclock_now());
614
615 *usec = yr > yl ? yr - yl : 0;
616 return 0;
617 }
618
619 case SOURCE_MESSAGE_POST: {
620 pa_memchunk c;
621
622 pa_mcalign_push(u->mcalign, chunk);
623
624 while (pa_mcalign_pop(u->mcalign, &c) >= 0) {
625
626 if (PA_SOURCE_IS_OPENED(u->source->thread_info.state))
627 pa_source_post(u->source, &c);
628
629 pa_memblock_unref(c.memblock);
630
631 u->counter += (int64_t) c.length;
632 }
633
634 return 0;
635 }
636
637 case SOURCE_MESSAGE_REMOTE_SUSPEND:
638
639 stream_suspend_within_thread(u, !!PA_PTR_TO_UINT(data));
640 return 0;
641
642 case SOURCE_MESSAGE_UPDATE_LATENCY: {
643 pa_usec_t y;
644
645 y = pa_bytes_to_usec((uint64_t) u->counter, &u->source->sample_spec);
646 y += (pa_usec_t) offset;
647
648 pa_smoother_put(u->smoother, pa_rtclock_now(), y);
649
650 /* We can access this freely here, since the main thread is waiting for us */
651 u->thread_transport_usec = u->transport_usec;
652
653 return 0;
654 }
655 }
656
657 return pa_source_process_msg(o, code, data, offset, chunk);
658 }
659
660 /* Called from main context */
661 static int source_set_state(pa_source *s, pa_source_state_t state) {
662 struct userdata *u;
663 pa_source_assert_ref(s);
664 u = s->userdata;
665
666 switch ((pa_source_state_t) state) {
667
668 case PA_SOURCE_SUSPENDED:
669 pa_assert(PA_SOURCE_IS_OPENED(s->state));
670 stream_cork(u, TRUE);
671 break;
672
673 case PA_SOURCE_IDLE:
674 case PA_SOURCE_RUNNING:
675 if (s->state == PA_SOURCE_SUSPENDED)
676 stream_cork(u, FALSE);
677 break;
678
679 case PA_SOURCE_UNLINKED:
680 case PA_SOURCE_INIT:
681 case PA_SINK_INVALID_STATE:
682 ;
683 }
684
685 return 0;
686 }
687
688 #endif
689
690 static void thread_func(void *userdata) {
691 struct userdata *u = userdata;
692
693 pa_assert(u);
694
695 pa_log_debug("Thread starting up");
696
697 pa_thread_mq_install(&u->thread_mq);
698
699 for (;;) {
700 int ret;
701
702 #ifdef TUNNEL_SINK
703 if (PA_SINK_IS_OPENED(u->sink->thread_info.state))
704 if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
705 pa_sink_process_rewind(u->sink, 0);
706 #endif
707
708 if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
709 goto fail;
710
711 if (ret == 0)
712 goto finish;
713 }
714
715 fail:
716 /* If this was no regular exit from the loop we have to continue
717 * processing messages until we received PA_MESSAGE_SHUTDOWN */
718 pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
719 pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
720
721 finish:
722 pa_log_debug("Thread shutting down");
723 }
724
725 #ifdef TUNNEL_SINK
726 /* Called from main context */
727 static void command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
728 struct userdata *u = userdata;
729 uint32_t bytes, channel;
730
731 pa_assert(pd);
732 pa_assert(command == PA_COMMAND_REQUEST);
733 pa_assert(t);
734 pa_assert(u);
735 pa_assert(u->pdispatch == pd);
736
737 if (pa_tagstruct_getu32(t, &channel) < 0 ||
738 pa_tagstruct_getu32(t, &bytes) < 0) {
739 pa_log("Invalid protocol reply");
740 goto fail;
741 }
742
743 if (channel != u->channel) {
744 pa_log("Received data for invalid channel");
745 goto fail;
746 }
747
748 pa_asyncmsgq_post(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REQUEST, NULL, bytes, NULL, NULL);
749 return;
750
751 fail:
752 pa_module_unload_request(u->module, TRUE);
753 }
754
755 #endif
756
757 /* Called from main context */
758 static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
759 struct userdata *u = userdata;
760 pa_usec_t sink_usec, source_usec;
761 pa_bool_t playing;
762 int64_t write_index, read_index;
763 struct timeval local, remote, now;
764 pa_sample_spec *ss;
765 int64_t delay;
766
767 pa_assert(pd);
768 pa_assert(u);
769
770 if (command != PA_COMMAND_REPLY) {
771 if (command == PA_COMMAND_ERROR)
772 pa_log("Failed to get latency.");
773 else
774 pa_log("Protocol error.");
775 goto fail;
776 }
777
778 if (pa_tagstruct_get_usec(t, &sink_usec) < 0 ||
779 pa_tagstruct_get_usec(t, &source_usec) < 0 ||
780 pa_tagstruct_get_boolean(t, &playing) < 0 ||
781 pa_tagstruct_get_timeval(t, &local) < 0 ||
782 pa_tagstruct_get_timeval(t, &remote) < 0 ||
783 pa_tagstruct_gets64(t, &write_index) < 0 ||
784 pa_tagstruct_gets64(t, &read_index) < 0) {
785 pa_log("Invalid reply.");
786 goto fail;
787 }
788
789 #ifdef TUNNEL_SINK
790 if (u->version >= 13) {
791 uint64_t underrun_for = 0, playing_for = 0;
792
793 if (pa_tagstruct_getu64(t, &underrun_for) < 0 ||
794 pa_tagstruct_getu64(t, &playing_for) < 0) {
795 pa_log("Invalid reply.");
796 goto fail;
797 }
798 }
799 #endif
800
801 if (!pa_tagstruct_eof(t)) {
802 pa_log("Invalid reply.");
803 goto fail;
804 }
805
806 if (tag < u->ignore_latency_before) {
807 return;
808 }
809
810 pa_gettimeofday(&now);
811
812 /* Calculate transport usec */
813 if (pa_timeval_cmp(&local, &remote) < 0 && pa_timeval_cmp(&remote, &now)) {
814 /* local and remote seem to have synchronized clocks */
815 #ifdef TUNNEL_SINK
816 u->transport_usec = pa_timeval_diff(&remote, &local);
817 #else
818 u->transport_usec = pa_timeval_diff(&now, &remote);
819 #endif
820 } else
821 u->transport_usec = pa_timeval_diff(&now, &local)/2;
822
823 /* First, take the device's delay */
824 #ifdef TUNNEL_SINK
825 delay = (int64_t) sink_usec;
826 ss = &u->sink->sample_spec;
827 #else
828 delay = (int64_t) source_usec;
829 ss = &u->source->sample_spec;
830 #endif
831
832 /* Add the length of our server-side buffer */
833 if (write_index >= read_index)
834 delay += (int64_t) pa_bytes_to_usec((uint64_t) (write_index-read_index), ss);
835 else
836 delay -= (int64_t) pa_bytes_to_usec((uint64_t) (read_index-write_index), ss);
837
838 /* Our measurements are already out of date, hence correct by the *
839 * transport latency */
840 #ifdef TUNNEL_SINK
841 delay -= (int64_t) u->transport_usec;
842 #else
843 delay += (int64_t) u->transport_usec;
844 #endif
845
846 /* Now correct by what we have have read/written since we requested the update */
847 #ifdef TUNNEL_SINK
848 delay += (int64_t) pa_bytes_to_usec((uint64_t) u->counter_delta, ss);
849 #else
850 delay -= (int64_t) pa_bytes_to_usec((uint64_t) u->counter_delta, ss);
851 #endif
852
853 #ifdef TUNNEL_SINK
854 pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UPDATE_LATENCY, 0, delay, NULL);
855 #else
856 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_UPDATE_LATENCY, 0, delay, NULL);
857 #endif
858
859 return;
860
861 fail:
862
863 pa_module_unload_request(u->module, TRUE);
864 }
865
866 /* Called from main context */
867 static void request_latency(struct userdata *u) {
868 pa_tagstruct *t;
869 struct timeval now;
870 uint32_t tag;
871 pa_assert(u);
872
873 t = pa_tagstruct_new(NULL, 0);
874 #ifdef TUNNEL_SINK
875 pa_tagstruct_putu32(t, PA_COMMAND_GET_PLAYBACK_LATENCY);
876 #else
877 pa_tagstruct_putu32(t, PA_COMMAND_GET_RECORD_LATENCY);
878 #endif
879 pa_tagstruct_putu32(t, tag = u->ctag++);
880 pa_tagstruct_putu32(t, u->channel);
881
882 pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
883
884 pa_pstream_send_tagstruct(u->pstream, t);
885 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, u, NULL);
886
887 u->ignore_latency_before = tag;
888 u->counter_delta = 0;
889 }
890
891 /* Called from main context */
892 static void timeout_callback(pa_mainloop_api *m, pa_time_event *e, const struct timeval *t, void *userdata) {
893 struct userdata *u = userdata;
894
895 pa_assert(m);
896 pa_assert(e);
897 pa_assert(u);
898
899 request_latency(u);
900
901 pa_core_rttime_restart(u->core, e, pa_rtclock_now() + LATENCY_INTERVAL);
902 }
903
904 /* Called from main context */
905 static void update_description(struct userdata *u) {
906 char *d;
907 char un[128], hn[128];
908 pa_tagstruct *t;
909
910 pa_assert(u);
911
912 if (!u->server_fqdn || !u->user_name || !u->device_description)
913 return;
914
915 d = pa_sprintf_malloc("%s on %s@%s", u->device_description, u->user_name, u->server_fqdn);
916
917 #ifdef TUNNEL_SINK
918 pa_sink_set_description(u->sink, d);
919 pa_proplist_sets(u->sink->proplist, "tunnel.remote.user", u->user_name);
920 pa_proplist_sets(u->sink->proplist, "tunnel.remote.fqdn", u->server_fqdn);
921 pa_proplist_sets(u->sink->proplist, "tunnel.remote.description", u->device_description);
922 #else
923 pa_source_set_description(u->source, d);
924 pa_proplist_sets(u->source->proplist, "tunnel.remote.user", u->user_name);
925 pa_proplist_sets(u->source->proplist, "tunnel.remote.fqdn", u->server_fqdn);
926 pa_proplist_sets(u->source->proplist, "tunnel.remote.description", u->device_description);
927 #endif
928
929 pa_xfree(d);
930
931 d = pa_sprintf_malloc("%s for %s@%s", u->device_description,
932 pa_get_user_name(un, sizeof(un)),
933 pa_get_host_name(hn, sizeof(hn)));
934
935 t = pa_tagstruct_new(NULL, 0);
936 #ifdef TUNNEL_SINK
937 pa_tagstruct_putu32(t, PA_COMMAND_SET_PLAYBACK_STREAM_NAME);
938 #else
939 pa_tagstruct_putu32(t, PA_COMMAND_SET_RECORD_STREAM_NAME);
940 #endif
941 pa_tagstruct_putu32(t, u->ctag++);
942 pa_tagstruct_putu32(t, u->channel);
943 pa_tagstruct_puts(t, d);
944 pa_pstream_send_tagstruct(u->pstream, t);
945
946 pa_xfree(d);
947 }
948
949 /* Called from main context */
950 static void server_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
951 struct userdata *u = userdata;
952 pa_sample_spec ss;
953 pa_channel_map cm;
954 const char *server_name, *server_version, *user_name, *host_name, *default_sink_name, *default_source_name;
955 uint32_t cookie;
956
957 pa_assert(pd);
958 pa_assert(u);
959
960 if (command != PA_COMMAND_REPLY) {
961 if (command == PA_COMMAND_ERROR)
962 pa_log("Failed to get info.");
963 else
964 pa_log("Protocol error.");
965 goto fail;
966 }
967
968 if (pa_tagstruct_gets(t, &server_name) < 0 ||
969 pa_tagstruct_gets(t, &server_version) < 0 ||
970 pa_tagstruct_gets(t, &user_name) < 0 ||
971 pa_tagstruct_gets(t, &host_name) < 0 ||
972 pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
973 pa_tagstruct_gets(t, &default_sink_name) < 0 ||
974 pa_tagstruct_gets(t, &default_source_name) < 0 ||
975 pa_tagstruct_getu32(t, &cookie) < 0 ||
976 (u->version >= 15 &&
977 pa_tagstruct_get_channel_map(t, &cm) < 0)) {
978
979 pa_log("Parse failure");
980 goto fail;
981 }
982
983 if (!pa_tagstruct_eof(t)) {
984 pa_log("Packet too long");
985 goto fail;
986 }
987
988 pa_xfree(u->server_fqdn);
989 u->server_fqdn = pa_xstrdup(host_name);
990
991 pa_xfree(u->user_name);
992 u->user_name = pa_xstrdup(user_name);
993
994 update_description(u);
995
996 return;
997
998 fail:
999 pa_module_unload_request(u->module, TRUE);
1000 }
1001
1002 #ifdef TUNNEL_SINK
1003
1004 /* Called from main context */
1005 static void sink_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1006 struct userdata *u = userdata;
1007 uint32_t idx, owner_module, monitor_source, flags;
1008 const char *name, *description, *monitor_source_name, *driver;
1009 pa_sample_spec ss;
1010 pa_channel_map cm;
1011 pa_cvolume volume;
1012 pa_bool_t mute;
1013 pa_usec_t latency;
1014 pa_proplist *pl;
1015
1016 pa_assert(pd);
1017 pa_assert(u);
1018
1019 pl = pa_proplist_new();
1020
1021 if (command != PA_COMMAND_REPLY) {
1022 if (command == PA_COMMAND_ERROR)
1023 pa_log("Failed to get info.");
1024 else
1025 pa_log("Protocol error.");
1026 goto fail;
1027 }
1028
1029 if (pa_tagstruct_getu32(t, &idx) < 0 ||
1030 pa_tagstruct_gets(t, &name) < 0 ||
1031 pa_tagstruct_gets(t, &description) < 0 ||
1032 pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1033 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1034 pa_tagstruct_getu32(t, &owner_module) < 0 ||
1035 pa_tagstruct_get_cvolume(t, &volume) < 0 ||
1036 pa_tagstruct_get_boolean(t, &mute) < 0 ||
1037 pa_tagstruct_getu32(t, &monitor_source) < 0 ||
1038 pa_tagstruct_gets(t, &monitor_source_name) < 0 ||
1039 pa_tagstruct_get_usec(t, &latency) < 0 ||
1040 pa_tagstruct_gets(t, &driver) < 0 ||
1041 pa_tagstruct_getu32(t, &flags) < 0) {
1042
1043 pa_log("Parse failure");
1044 goto fail;
1045 }
1046
1047 if (u->version >= 13) {
1048 pa_usec_t configured_latency;
1049
1050 if (pa_tagstruct_get_proplist(t, pl) < 0 ||
1051 pa_tagstruct_get_usec(t, &configured_latency) < 0) {
1052
1053 pa_log("Parse failure");
1054 goto fail;
1055 }
1056 }
1057
1058 if (u->version >= 15) {
1059 pa_volume_t base_volume;
1060 uint32_t state, n_volume_steps, card;
1061
1062 if (pa_tagstruct_get_volume(t, &base_volume) < 0 ||
1063 pa_tagstruct_getu32(t, &state) < 0 ||
1064 pa_tagstruct_getu32(t, &n_volume_steps) < 0 ||
1065 pa_tagstruct_getu32(t, &card) < 0) {
1066
1067 pa_log("Parse failure");
1068 goto fail;
1069 }
1070 }
1071
1072 if (!pa_tagstruct_eof(t)) {
1073 pa_log("Packet too long");
1074 goto fail;
1075 }
1076
1077 pa_proplist_free(pl);
1078
1079 if (!u->sink_name || strcmp(name, u->sink_name))
1080 return;
1081
1082 pa_xfree(u->device_description);
1083 u->device_description = pa_xstrdup(description);
1084
1085 update_description(u);
1086
1087 return;
1088
1089 fail:
1090 pa_module_unload_request(u->module, TRUE);
1091 pa_proplist_free(pl);
1092 }
1093
1094 /* Called from main context */
1095 static void sink_input_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1096 struct userdata *u = userdata;
1097 uint32_t idx, owner_module, client, sink;
1098 pa_usec_t buffer_usec, sink_usec;
1099 const char *name, *driver, *resample_method;
1100 pa_bool_t mute;
1101 pa_sample_spec sample_spec;
1102 pa_channel_map channel_map;
1103 pa_cvolume volume;
1104 pa_proplist *pl;
1105
1106 pa_assert(pd);
1107 pa_assert(u);
1108
1109 pl = pa_proplist_new();
1110
1111 if (command != PA_COMMAND_REPLY) {
1112 if (command == PA_COMMAND_ERROR)
1113 pa_log("Failed to get info.");
1114 else
1115 pa_log("Protocol error.");
1116 goto fail;
1117 }
1118
1119 if (pa_tagstruct_getu32(t, &idx) < 0 ||
1120 pa_tagstruct_gets(t, &name) < 0 ||
1121 pa_tagstruct_getu32(t, &owner_module) < 0 ||
1122 pa_tagstruct_getu32(t, &client) < 0 ||
1123 pa_tagstruct_getu32(t, &sink) < 0 ||
1124 pa_tagstruct_get_sample_spec(t, &sample_spec) < 0 ||
1125 pa_tagstruct_get_channel_map(t, &channel_map) < 0 ||
1126 pa_tagstruct_get_cvolume(t, &volume) < 0 ||
1127 pa_tagstruct_get_usec(t, &buffer_usec) < 0 ||
1128 pa_tagstruct_get_usec(t, &sink_usec) < 0 ||
1129 pa_tagstruct_gets(t, &resample_method) < 0 ||
1130 pa_tagstruct_gets(t, &driver) < 0) {
1131
1132 pa_log("Parse failure");
1133 goto fail;
1134 }
1135
1136 if (u->version >= 11) {
1137 if (pa_tagstruct_get_boolean(t, &mute) < 0) {
1138
1139 pa_log("Parse failure");
1140 goto fail;
1141 }
1142 }
1143
1144 if (u->version >= 13) {
1145 if (pa_tagstruct_get_proplist(t, pl) < 0) {
1146
1147 pa_log("Parse failure");
1148 goto fail;
1149 }
1150 }
1151
1152 if (!pa_tagstruct_eof(t)) {
1153 pa_log("Packet too long");
1154 goto fail;
1155 }
1156
1157 pa_proplist_free(pl);
1158
1159 if (idx != u->device_index)
1160 return;
1161
1162 pa_assert(u->sink);
1163
1164 if ((u->version < 11 || !!mute == !!u->sink->muted) &&
1165 pa_cvolume_equal(&volume, &u->sink->real_volume))
1166 return;
1167
1168 pa_sink_volume_changed(u->sink, &volume);
1169
1170 if (u->version >= 11)
1171 pa_sink_mute_changed(u->sink, mute);
1172
1173 return;
1174
1175 fail:
1176 pa_module_unload_request(u->module, TRUE);
1177 pa_proplist_free(pl);
1178 }
1179
1180 #else
1181
1182 /* Called from main context */
1183 static void source_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1184 struct userdata *u = userdata;
1185 uint32_t idx, owner_module, monitor_of_sink, flags;
1186 const char *name, *description, *monitor_of_sink_name, *driver;
1187 pa_sample_spec ss;
1188 pa_channel_map cm;
1189 pa_cvolume volume;
1190 pa_bool_t mute;
1191 pa_usec_t latency, configured_latency;
1192 pa_proplist *pl;
1193
1194 pa_assert(pd);
1195 pa_assert(u);
1196
1197 pl = pa_proplist_new();
1198
1199 if (command != PA_COMMAND_REPLY) {
1200 if (command == PA_COMMAND_ERROR)
1201 pa_log("Failed to get info.");
1202 else
1203 pa_log("Protocol error.");
1204 goto fail;
1205 }
1206
1207 if (pa_tagstruct_getu32(t, &idx) < 0 ||
1208 pa_tagstruct_gets(t, &name) < 0 ||
1209 pa_tagstruct_gets(t, &description) < 0 ||
1210 pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1211 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1212 pa_tagstruct_getu32(t, &owner_module) < 0 ||
1213 pa_tagstruct_get_cvolume(t, &volume) < 0 ||
1214 pa_tagstruct_get_boolean(t, &mute) < 0 ||
1215 pa_tagstruct_getu32(t, &monitor_of_sink) < 0 ||
1216 pa_tagstruct_gets(t, &monitor_of_sink_name) < 0 ||
1217 pa_tagstruct_get_usec(t, &latency) < 0 ||
1218 pa_tagstruct_gets(t, &driver) < 0 ||
1219 pa_tagstruct_getu32(t, &flags) < 0) {
1220
1221 pa_log("Parse failure");
1222 goto fail;
1223 }
1224
1225 if (u->version >= 13) {
1226 if (pa_tagstruct_get_proplist(t, pl) < 0 ||
1227 pa_tagstruct_get_usec(t, &configured_latency) < 0) {
1228
1229 pa_log("Parse failure");
1230 goto fail;
1231 }
1232 }
1233
1234 if (u->version >= 15) {
1235 pa_volume_t base_volume;
1236 uint32_t state, n_volume_steps, card;
1237
1238 if (pa_tagstruct_get_volume(t, &base_volume) < 0 ||
1239 pa_tagstruct_getu32(t, &state) < 0 ||
1240 pa_tagstruct_getu32(t, &n_volume_steps) < 0 ||
1241 pa_tagstruct_getu32(t, &card) < 0) {
1242
1243 pa_log("Parse failure");
1244 goto fail;
1245 }
1246 }
1247
1248 if (!pa_tagstruct_eof(t)) {
1249 pa_log("Packet too long");
1250 goto fail;
1251 }
1252
1253 pa_proplist_free(pl);
1254
1255 if (!u->source_name || strcmp(name, u->source_name))
1256 return;
1257
1258 pa_xfree(u->device_description);
1259 u->device_description = pa_xstrdup(description);
1260
1261 update_description(u);
1262
1263 return;
1264
1265 fail:
1266 pa_module_unload_request(u->module, TRUE);
1267 pa_proplist_free(pl);
1268 }
1269
1270 #endif
1271
1272 /* Called from main context */
1273 static void request_info(struct userdata *u) {
1274 pa_tagstruct *t;
1275 uint32_t tag;
1276 pa_assert(u);
1277
1278 t = pa_tagstruct_new(NULL, 0);
1279 pa_tagstruct_putu32(t, PA_COMMAND_GET_SERVER_INFO);
1280 pa_tagstruct_putu32(t, tag = u->ctag++);
1281 pa_pstream_send_tagstruct(u->pstream, t);
1282 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, server_info_cb, u, NULL);
1283
1284 #ifdef TUNNEL_SINK
1285 t = pa_tagstruct_new(NULL, 0);
1286 pa_tagstruct_putu32(t, PA_COMMAND_GET_SINK_INPUT_INFO);
1287 pa_tagstruct_putu32(t, tag = u->ctag++);
1288 pa_tagstruct_putu32(t, u->device_index);
1289 pa_pstream_send_tagstruct(u->pstream, t);
1290 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, sink_input_info_cb, u, NULL);
1291
1292 if (u->sink_name) {
1293 t = pa_tagstruct_new(NULL, 0);
1294 pa_tagstruct_putu32(t, PA_COMMAND_GET_SINK_INFO);
1295 pa_tagstruct_putu32(t, tag = u->ctag++);
1296 pa_tagstruct_putu32(t, PA_INVALID_INDEX);
1297 pa_tagstruct_puts(t, u->sink_name);
1298 pa_pstream_send_tagstruct(u->pstream, t);
1299 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, sink_info_cb, u, NULL);
1300 }
1301 #else
1302 if (u->source_name) {
1303 t = pa_tagstruct_new(NULL, 0);
1304 pa_tagstruct_putu32(t, PA_COMMAND_GET_SOURCE_INFO);
1305 pa_tagstruct_putu32(t, tag = u->ctag++);
1306 pa_tagstruct_putu32(t, PA_INVALID_INDEX);
1307 pa_tagstruct_puts(t, u->source_name);
1308 pa_pstream_send_tagstruct(u->pstream, t);
1309 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, source_info_cb, u, NULL);
1310 }
1311 #endif
1312 }
1313
1314 /* Called from main context */
1315 static void command_subscribe_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1316 struct userdata *u = userdata;
1317 pa_subscription_event_type_t e;
1318 uint32_t idx;
1319
1320 pa_assert(pd);
1321 pa_assert(t);
1322 pa_assert(u);
1323 pa_assert(command == PA_COMMAND_SUBSCRIBE_EVENT);
1324
1325 if (pa_tagstruct_getu32(t, &e) < 0 ||
1326 pa_tagstruct_getu32(t, &idx) < 0) {
1327 pa_log("Invalid protocol reply");
1328 pa_module_unload_request(u->module, TRUE);
1329 return;
1330 }
1331
1332 if (e != (PA_SUBSCRIPTION_EVENT_SERVER|PA_SUBSCRIPTION_EVENT_CHANGE) &&
1333 #ifdef TUNNEL_SINK
1334 e != (PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE) &&
1335 e != (PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE)
1336 #else
1337 e != (PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE)
1338 #endif
1339 )
1340 return;
1341
1342 request_info(u);
1343 }
1344
1345 /* Called from main context */
1346 static void start_subscribe(struct userdata *u) {
1347 pa_tagstruct *t;
1348 uint32_t tag;
1349 pa_assert(u);
1350
1351 t = pa_tagstruct_new(NULL, 0);
1352 pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE);
1353 pa_tagstruct_putu32(t, tag = u->ctag++);
1354 pa_tagstruct_putu32(t, PA_SUBSCRIPTION_MASK_SERVER|
1355 #ifdef TUNNEL_SINK
1356 PA_SUBSCRIPTION_MASK_SINK_INPUT|PA_SUBSCRIPTION_MASK_SINK
1357 #else
1358 PA_SUBSCRIPTION_MASK_SOURCE
1359 #endif
1360 );
1361
1362 pa_pstream_send_tagstruct(u->pstream, t);
1363 }
1364
1365 /* Called from main context */
1366 static void create_stream_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1367 struct userdata *u = userdata;
1368 #ifdef TUNNEL_SINK
1369 uint32_t bytes;
1370 #endif
1371
1372 pa_assert(pd);
1373 pa_assert(u);
1374 pa_assert(u->pdispatch == pd);
1375
1376 if (command != PA_COMMAND_REPLY) {
1377 if (command == PA_COMMAND_ERROR)
1378 pa_log("Failed to create stream.");
1379 else
1380 pa_log("Protocol error.");
1381 goto fail;
1382 }
1383
1384 if (pa_tagstruct_getu32(t, &u->channel) < 0 ||
1385 pa_tagstruct_getu32(t, &u->device_index) < 0
1386 #ifdef TUNNEL_SINK
1387 || pa_tagstruct_getu32(t, &bytes) < 0
1388 #endif
1389 )
1390 goto parse_error;
1391
1392 if (u->version >= 9) {
1393 #ifdef TUNNEL_SINK
1394 if (pa_tagstruct_getu32(t, &u->maxlength) < 0 ||
1395 pa_tagstruct_getu32(t, &u->tlength) < 0 ||
1396 pa_tagstruct_getu32(t, &u->prebuf) < 0 ||
1397 pa_tagstruct_getu32(t, &u->minreq) < 0)
1398 goto parse_error;
1399 #else
1400 if (pa_tagstruct_getu32(t, &u->maxlength) < 0 ||
1401 pa_tagstruct_getu32(t, &u->fragsize) < 0)
1402 goto parse_error;
1403 #endif
1404 }
1405
1406 if (u->version >= 12) {
1407 pa_sample_spec ss;
1408 pa_channel_map cm;
1409 uint32_t device_index;
1410 const char *dn;
1411 pa_bool_t suspended;
1412
1413 if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1414 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1415 pa_tagstruct_getu32(t, &device_index) < 0 ||
1416 pa_tagstruct_gets(t, &dn) < 0 ||
1417 pa_tagstruct_get_boolean(t, &suspended) < 0)
1418 goto parse_error;
1419
1420 #ifdef TUNNEL_SINK
1421 pa_xfree(u->sink_name);
1422 u->sink_name = pa_xstrdup(dn);
1423 #else
1424 pa_xfree(u->source_name);
1425 u->source_name = pa_xstrdup(dn);
1426 #endif
1427 }
1428
1429 if (u->version >= 13) {
1430 pa_usec_t usec;
1431
1432 if (pa_tagstruct_get_usec(t, &usec) < 0)
1433 goto parse_error;
1434
1435 /* #ifdef TUNNEL_SINK */
1436 /* pa_sink_set_latency_range(u->sink, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1437 /* #else */
1438 /* pa_source_set_latency_range(u->source, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1439 /* #endif */
1440 }
1441
1442 if (!pa_tagstruct_eof(t))
1443 goto parse_error;
1444
1445 start_subscribe(u);
1446 request_info(u);
1447
1448 pa_assert(!u->time_event);
1449 u->time_event = pa_core_rttime_new(u->core, pa_rtclock_now() + LATENCY_INTERVAL, timeout_callback, u);
1450
1451 request_latency(u);
1452
1453 pa_log_debug("Stream created.");
1454
1455 #ifdef TUNNEL_SINK
1456 pa_asyncmsgq_post(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REQUEST, NULL, bytes, NULL, NULL);
1457 #endif
1458
1459 return;
1460
1461 parse_error:
1462 pa_log("Invalid reply. (Create stream)");
1463
1464 fail:
1465 pa_module_unload_request(u->module, TRUE);
1466
1467 }
1468
1469 /* Called from main context */
1470 static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1471 struct userdata *u = userdata;
1472 pa_tagstruct *reply;
1473 char name[256], un[128], hn[128];
1474 #ifdef TUNNEL_SINK
1475 pa_cvolume volume;
1476 #endif
1477
1478 pa_assert(pd);
1479 pa_assert(u);
1480 pa_assert(u->pdispatch == pd);
1481
1482 if (command != PA_COMMAND_REPLY ||
1483 pa_tagstruct_getu32(t, &u->version) < 0 ||
1484 !pa_tagstruct_eof(t)) {
1485
1486 if (command == PA_COMMAND_ERROR)
1487 pa_log("Failed to authenticate");
1488 else
1489 pa_log("Protocol error.");
1490
1491 goto fail;
1492 }
1493
1494 /* Minimum supported protocol version */
1495 if (u->version < 8) {
1496 pa_log("Incompatible protocol version");
1497 goto fail;
1498 }
1499
1500 /* Starting with protocol version 13 the MSB of the version tag
1501 reflects if shm is enabled for this connection or not. We don't
1502 support SHM here at all, so we just ignore this. */
1503
1504 if (u->version >= 13)
1505 u->version &= 0x7FFFFFFFU;
1506
1507 pa_log_debug("Protocol version: remote %u, local %u", u->version, PA_PROTOCOL_VERSION);
1508
1509 #ifdef TUNNEL_SINK
1510 pa_proplist_setf(u->sink->proplist, "tunnel.remote_version", "%u", u->version);
1511 pa_sink_update_proplist(u->sink, 0, NULL);
1512
1513 pa_snprintf(name, sizeof(name), "%s for %s@%s",
1514 u->sink_name,
1515 pa_get_user_name(un, sizeof(un)),
1516 pa_get_host_name(hn, sizeof(hn)));
1517 #else
1518 pa_proplist_setf(u->source->proplist, "tunnel.remote_version", "%u", u->version);
1519 pa_source_update_proplist(u->source, 0, NULL);
1520
1521 pa_snprintf(name, sizeof(name), "%s for %s@%s",
1522 u->source_name,
1523 pa_get_user_name(un, sizeof(un)),
1524 pa_get_host_name(hn, sizeof(hn)));
1525 #endif
1526
1527 reply = pa_tagstruct_new(NULL, 0);
1528 pa_tagstruct_putu32(reply, PA_COMMAND_SET_CLIENT_NAME);
1529 pa_tagstruct_putu32(reply, tag = u->ctag++);
1530
1531 if (u->version >= 13) {
1532 pa_proplist *pl;
1533 pl = pa_proplist_new();
1534 pa_proplist_sets(pl, PA_PROP_APPLICATION_ID, "org.PulseAudio.PulseAudio");
1535 pa_proplist_sets(pl, PA_PROP_APPLICATION_VERSION, PACKAGE_VERSION);
1536 pa_init_proplist(pl);
1537 pa_tagstruct_put_proplist(reply, pl);
1538 pa_proplist_free(pl);
1539 } else
1540 pa_tagstruct_puts(reply, "PulseAudio");
1541
1542 pa_pstream_send_tagstruct(u->pstream, reply);
1543 /* We ignore the server's reply here */
1544
1545 reply = pa_tagstruct_new(NULL, 0);
1546
1547 if (u->version < 13)
1548 /* Only for older PA versions we need to fill in the maxlength */
1549 u->maxlength = 4*1024*1024;
1550
1551 #ifdef TUNNEL_SINK
1552 u->tlength = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_TLENGTH_MSEC, &u->sink->sample_spec);
1553 u->minreq = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_MINREQ_MSEC, &u->sink->sample_spec);
1554 u->prebuf = u->tlength;
1555 #else
1556 u->fragsize = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_FRAGSIZE_MSEC, &u->source->sample_spec);
1557 #endif
1558
1559 #ifdef TUNNEL_SINK
1560 pa_tagstruct_putu32(reply, PA_COMMAND_CREATE_PLAYBACK_STREAM);
1561 pa_tagstruct_putu32(reply, tag = u->ctag++);
1562
1563 if (u->version < 13)
1564 pa_tagstruct_puts(reply, name);
1565
1566 pa_tagstruct_put_sample_spec(reply, &u->sink->sample_spec);
1567 pa_tagstruct_put_channel_map(reply, &u->sink->channel_map);
1568 pa_tagstruct_putu32(reply, PA_INVALID_INDEX);
1569 pa_tagstruct_puts(reply, u->sink_name);
1570 pa_tagstruct_putu32(reply, u->maxlength);
1571 pa_tagstruct_put_boolean(reply, !PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)));
1572 pa_tagstruct_putu32(reply, u->tlength);
1573 pa_tagstruct_putu32(reply, u->prebuf);
1574 pa_tagstruct_putu32(reply, u->minreq);
1575 pa_tagstruct_putu32(reply, 0);
1576 pa_cvolume_reset(&volume, u->sink->sample_spec.channels);
1577 pa_tagstruct_put_cvolume(reply, &volume);
1578 #else
1579 pa_tagstruct_putu32(reply, PA_COMMAND_CREATE_RECORD_STREAM);
1580 pa_tagstruct_putu32(reply, tag = u->ctag++);
1581
1582 if (u->version < 13)
1583 pa_tagstruct_puts(reply, name);
1584
1585 pa_tagstruct_put_sample_spec(reply, &u->source->sample_spec);
1586 pa_tagstruct_put_channel_map(reply, &u->source->channel_map);
1587 pa_tagstruct_putu32(reply, PA_INVALID_INDEX);
1588 pa_tagstruct_puts(reply, u->source_name);
1589 pa_tagstruct_putu32(reply, u->maxlength);
1590 pa_tagstruct_put_boolean(reply, !PA_SOURCE_IS_OPENED(pa_source_get_state(u->source)));
1591 pa_tagstruct_putu32(reply, u->fragsize);
1592 #endif
1593
1594 if (u->version >= 12) {
1595 pa_tagstruct_put_boolean(reply, FALSE); /* no_remap */
1596 pa_tagstruct_put_boolean(reply, FALSE); /* no_remix */
1597 pa_tagstruct_put_boolean(reply, FALSE); /* fix_format */
1598 pa_tagstruct_put_boolean(reply, FALSE); /* fix_rate */
1599 pa_tagstruct_put_boolean(reply, FALSE); /* fix_channels */
1600 pa_tagstruct_put_boolean(reply, TRUE); /* no_move */
1601 pa_tagstruct_put_boolean(reply, FALSE); /* variable_rate */
1602 }
1603
1604 if (u->version >= 13) {
1605 pa_proplist *pl;
1606
1607 pa_tagstruct_put_boolean(reply, FALSE); /* start muted/peak detect*/
1608 pa_tagstruct_put_boolean(reply, TRUE); /* adjust_latency */
1609
1610 pl = pa_proplist_new();
1611 pa_proplist_sets(pl, PA_PROP_MEDIA_NAME, name);
1612 pa_proplist_sets(pl, PA_PROP_MEDIA_ROLE, "abstract");
1613 pa_tagstruct_put_proplist(reply, pl);
1614 pa_proplist_free(pl);
1615
1616 #ifndef TUNNEL_SINK
1617 pa_tagstruct_putu32(reply, PA_INVALID_INDEX); /* direct on input */
1618 #endif
1619 }
1620
1621 if (u->version >= 14) {
1622 #ifdef TUNNEL_SINK
1623 pa_tagstruct_put_boolean(reply, FALSE); /* volume_set */
1624 #endif
1625 pa_tagstruct_put_boolean(reply, TRUE); /* early rquests */
1626 }
1627
1628 if (u->version >= 15) {
1629 #ifdef TUNNEL_SINK
1630 pa_tagstruct_put_boolean(reply, FALSE); /* muted_set */
1631 #endif
1632 pa_tagstruct_put_boolean(reply, FALSE); /* don't inhibit auto suspend */
1633 pa_tagstruct_put_boolean(reply, FALSE); /* fail on suspend */
1634 }
1635
1636 pa_pstream_send_tagstruct(u->pstream, reply);
1637 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, create_stream_callback, u, NULL);
1638
1639 pa_log_debug("Connection authenticated, creating stream ...");
1640
1641 return;
1642
1643 fail:
1644 pa_module_unload_request(u->module, TRUE);
1645 }
1646
1647 /* Called from main context */
1648 static void pstream_die_callback(pa_pstream *p, void *userdata) {
1649 struct userdata *u = userdata;
1650
1651 pa_assert(p);
1652 pa_assert(u);
1653
1654 pa_log_warn("Stream died.");
1655 pa_module_unload_request(u->module, TRUE);
1656 }
1657
1658 /* Called from main context */
1659 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
1660 struct userdata *u = userdata;
1661
1662 pa_assert(p);
1663 pa_assert(packet);
1664 pa_assert(u);
1665
1666 if (pa_pdispatch_run(u->pdispatch, packet, creds, u) < 0) {
1667 pa_log("Invalid packet");
1668 pa_module_unload_request(u->module, TRUE);
1669 return;
1670 }
1671 }
1672
1673 #ifndef TUNNEL_SINK
1674 /* Called from main context */
1675 static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata) {
1676 struct userdata *u = userdata;
1677
1678 pa_assert(p);
1679 pa_assert(chunk);
1680 pa_assert(u);
1681
1682 if (channel != u->channel) {
1683 pa_log("Received memory block on bad channel.");
1684 pa_module_unload_request(u->module, TRUE);
1685 return;
1686 }
1687
1688 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_POST, PA_UINT_TO_PTR(seek), offset, chunk);
1689
1690 u->counter_delta += (int64_t) chunk->length;
1691 }
1692 #endif
1693
1694 /* Called from main context */
1695 static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) {
1696 struct userdata *u = userdata;
1697 pa_tagstruct *t;
1698 uint32_t tag;
1699
1700 pa_assert(sc);
1701 pa_assert(u);
1702 pa_assert(u->client == sc);
1703
1704 pa_socket_client_unref(u->client);
1705 u->client = NULL;
1706
1707 if (!io) {
1708 pa_log("Connection failed: %s", pa_cstrerror(errno));
1709 pa_module_unload_request(u->module, TRUE);
1710 return;
1711 }
1712
1713 u->pstream = pa_pstream_new(u->core->mainloop, io, u->core->mempool);
1714 u->pdispatch = pa_pdispatch_new(u->core->mainloop, TRUE, command_table, PA_COMMAND_MAX);
1715
1716 pa_pstream_set_die_callback(u->pstream, pstream_die_callback, u);
1717 pa_pstream_set_recieve_packet_callback(u->pstream, pstream_packet_callback, u);
1718 #ifndef TUNNEL_SINK
1719 pa_pstream_set_recieve_memblock_callback(u->pstream, pstream_memblock_callback, u);
1720 #endif
1721
1722 t = pa_tagstruct_new(NULL, 0);
1723 pa_tagstruct_putu32(t, PA_COMMAND_AUTH);
1724 pa_tagstruct_putu32(t, tag = u->ctag++);
1725 pa_tagstruct_putu32(t, PA_PROTOCOL_VERSION);
1726
1727 pa_tagstruct_put_arbitrary(t, pa_auth_cookie_read(u->auth_cookie, PA_NATIVE_COOKIE_LENGTH), PA_NATIVE_COOKIE_LENGTH);
1728
1729 #ifdef HAVE_CREDS
1730 {
1731 pa_creds ucred;
1732
1733 if (pa_iochannel_creds_supported(io))
1734 pa_iochannel_creds_enable(io);
1735
1736 ucred.uid = getuid();
1737 ucred.gid = getgid();
1738
1739 pa_pstream_send_tagstruct_with_creds(u->pstream, t, &ucred);
1740 }
1741 #else
1742 pa_pstream_send_tagstruct(u->pstream, t);
1743 #endif
1744
1745 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, u, NULL);
1746
1747 pa_log_debug("Connection established, authenticating ...");
1748 }
1749
1750 #ifdef TUNNEL_SINK
1751
1752 /* Called from main context */
1753 static void sink_set_volume(pa_sink *sink) {
1754 struct userdata *u;
1755 pa_tagstruct *t;
1756 uint32_t tag;
1757
1758 pa_assert(sink);
1759 u = sink->userdata;
1760 pa_assert(u);
1761
1762 t = pa_tagstruct_new(NULL, 0);
1763 pa_tagstruct_putu32(t, PA_COMMAND_SET_SINK_INPUT_VOLUME);
1764 pa_tagstruct_putu32(t, tag = u->ctag++);
1765 pa_tagstruct_putu32(t, u->device_index);
1766 pa_tagstruct_put_cvolume(t, &sink->real_volume);
1767 pa_pstream_send_tagstruct(u->pstream, t);
1768 }
1769
1770 /* Called from main context */
1771 static void sink_set_mute(pa_sink *sink) {
1772 struct userdata *u;
1773 pa_tagstruct *t;
1774 uint32_t tag;
1775
1776 pa_assert(sink);
1777 u = sink->userdata;
1778 pa_assert(u);
1779
1780 if (u->version < 11)
1781 return;
1782
1783 t = pa_tagstruct_new(NULL, 0);
1784 pa_tagstruct_putu32(t, PA_COMMAND_SET_SINK_INPUT_MUTE);
1785 pa_tagstruct_putu32(t, tag = u->ctag++);
1786 pa_tagstruct_putu32(t, u->device_index);
1787 pa_tagstruct_put_boolean(t, !!sink->muted);
1788 pa_pstream_send_tagstruct(u->pstream, t);
1789 }
1790
1791 #endif
1792
1793 int pa__init(pa_module*m) {
1794 pa_modargs *ma = NULL;
1795 struct userdata *u = NULL;
1796 pa_sample_spec ss;
1797 pa_channel_map map;
1798 char *dn = NULL;
1799 #ifdef TUNNEL_SINK
1800 pa_sink_new_data data;
1801 #else
1802 pa_source_new_data data;
1803 #endif
1804
1805 pa_assert(m);
1806
1807 if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
1808 pa_log("Failed to parse module arguments");
1809 goto fail;
1810 }
1811
1812 m->userdata = u = pa_xnew0(struct userdata, 1);
1813 u->core = m->core;
1814 u->module = m;
1815 u->client = NULL;
1816 u->pdispatch = NULL;
1817 u->pstream = NULL;
1818 u->server_name = NULL;
1819 #ifdef TUNNEL_SINK
1820 u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));;
1821 u->sink = NULL;
1822 u->requested_bytes = 0;
1823 #else
1824 u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));;
1825 u->source = NULL;
1826 #endif
1827 u->smoother = pa_smoother_new(
1828 PA_USEC_PER_SEC,
1829 PA_USEC_PER_SEC*2,
1830 TRUE,
1831 TRUE,
1832 10,
1833 pa_rtclock_now(),
1834 FALSE);
1835 u->ctag = 1;
1836 u->device_index = u->channel = PA_INVALID_INDEX;
1837 u->time_event = NULL;
1838 u->ignore_latency_before = 0;
1839 u->transport_usec = u->thread_transport_usec = 0;
1840 u->remote_suspended = u->remote_corked = FALSE;
1841 u->counter = u->counter_delta = 0;
1842
1843 u->rtpoll = pa_rtpoll_new();
1844 pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
1845
1846 if (!(u->auth_cookie = pa_auth_cookie_get(u->core, pa_modargs_get_value(ma, "cookie", PA_NATIVE_COOKIE_FILE), PA_NATIVE_COOKIE_LENGTH)))
1847 goto fail;
1848
1849 if (!(u->server_name = pa_xstrdup(pa_modargs_get_value(ma, "server", NULL)))) {
1850 pa_log("No server specified.");
1851 goto fail;
1852 }
1853
1854 ss = m->core->default_sample_spec;
1855 map = m->core->default_channel_map;
1856 if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
1857 pa_log("Invalid sample format specification");
1858 goto fail;
1859 }
1860
1861 if (!(u->client = pa_socket_client_new_string(m->core->mainloop, TRUE, u->server_name, PA_NATIVE_DEFAULT_PORT))) {
1862 pa_log("Failed to connect to server '%s'", u->server_name);
1863 goto fail;
1864 }
1865
1866 pa_socket_client_set_callback(u->client, on_connection, u);
1867
1868 #ifdef TUNNEL_SINK
1869
1870 if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL))))
1871 dn = pa_sprintf_malloc("tunnel.%s", u->server_name);
1872
1873 pa_sink_new_data_init(&data);
1874 data.driver = __FILE__;
1875 data.module = m;
1876 data.namereg_fail = TRUE;
1877 pa_sink_new_data_set_name(&data, dn);
1878 pa_sink_new_data_set_sample_spec(&data, &ss);
1879 pa_sink_new_data_set_channel_map(&data, &map);
1880 pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->sink_name), u->sink_name ? " on " : "", u->server_name);
1881 pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
1882 if (u->sink_name)
1883 pa_proplist_sets(data.proplist, "tunnel.remote.sink", u->sink_name);
1884
1885 if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
1886 pa_log("Invalid properties");
1887 pa_sink_new_data_done(&data);
1888 goto fail;
1889 }
1890
1891 u->sink = pa_sink_new(m->core, &data, PA_SINK_NETWORK|PA_SINK_LATENCY|PA_SINK_HW_VOLUME_CTRL|PA_SINK_HW_MUTE_CTRL);
1892 pa_sink_new_data_done(&data);
1893
1894 if (!u->sink) {
1895 pa_log("Failed to create sink.");
1896 goto fail;
1897 }
1898
1899 u->sink->parent.process_msg = sink_process_msg;
1900 u->sink->userdata = u;
1901 u->sink->set_state = sink_set_state;
1902 u->sink->set_volume = sink_set_volume;
1903 u->sink->set_mute = sink_set_mute;
1904
1905 u->sink->refresh_volume = u->sink->refresh_muted = FALSE;
1906
1907 /* pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
1908
1909 pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
1910 pa_sink_set_rtpoll(u->sink, u->rtpoll);
1911
1912 #else
1913
1914 if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL))))
1915 dn = pa_sprintf_malloc("tunnel.%s", u->server_name);
1916
1917 pa_source_new_data_init(&data);
1918 data.driver = __FILE__;
1919 data.module = m;
1920 data.namereg_fail = TRUE;
1921 pa_source_new_data_set_name(&data, dn);
1922 pa_source_new_data_set_sample_spec(&data, &ss);
1923 pa_source_new_data_set_channel_map(&data, &map);
1924 pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->source_name), u->source_name ? " on " : "", u->server_name);
1925 pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
1926 if (u->source_name)
1927 pa_proplist_sets(data.proplist, "tunnel.remote.source", u->source_name);
1928
1929 if (pa_modargs_get_proplist(ma, "source_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
1930 pa_log("Invalid properties");
1931 pa_source_new_data_done(&data);
1932 goto fail;
1933 }
1934
1935 u->source = pa_source_new(m->core, &data, PA_SOURCE_NETWORK|PA_SOURCE_LATENCY);
1936 pa_source_new_data_done(&data);
1937
1938 if (!u->source) {
1939 pa_log("Failed to create source.");
1940 goto fail;
1941 }
1942
1943 u->source->parent.process_msg = source_process_msg;
1944 u->source->set_state = source_set_state;
1945 u->source->userdata = u;
1946
1947 /* pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
1948
1949 pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
1950 pa_source_set_rtpoll(u->source, u->rtpoll);
1951
1952 u->mcalign = pa_mcalign_new(pa_frame_size(&u->source->sample_spec));
1953 #endif
1954
1955 pa_xfree(dn);
1956
1957 u->time_event = NULL;
1958
1959 u->maxlength = (uint32_t) -1;
1960 #ifdef TUNNEL_SINK
1961 u->tlength = u->minreq = u->prebuf = (uint32_t) -1;
1962 #else
1963 u->fragsize = (uint32_t) -1;
1964 #endif
1965
1966 if (!(u->thread = pa_thread_new(thread_func, u))) {
1967 pa_log("Failed to create thread.");
1968 goto fail;
1969 }
1970
1971 #ifdef TUNNEL_SINK
1972 pa_sink_put(u->sink);
1973 #else
1974 pa_source_put(u->source);
1975 #endif
1976
1977 pa_modargs_free(ma);
1978
1979 return 0;
1980
1981 fail:
1982 pa__done(m);
1983
1984 if (ma)
1985 pa_modargs_free(ma);
1986
1987 pa_xfree(dn);
1988
1989 return -1;
1990 }
1991
1992 void pa__done(pa_module*m) {
1993 struct userdata* u;
1994
1995 pa_assert(m);
1996
1997 if (!(u = m->userdata))
1998 return;
1999
2000 #ifdef TUNNEL_SINK
2001 if (u->sink)
2002 pa_sink_unlink(u->sink);
2003 #else
2004 if (u->source)
2005 pa_source_unlink(u->source);
2006 #endif
2007
2008 if (u->thread) {
2009 pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
2010 pa_thread_free(u->thread);
2011 }
2012
2013 pa_thread_mq_done(&u->thread_mq);
2014
2015 #ifdef TUNNEL_SINK
2016 if (u->sink)
2017 pa_sink_unref(u->sink);
2018 #else
2019 if (u->source)
2020 pa_source_unref(u->source);
2021 #endif
2022
2023 if (u->rtpoll)
2024 pa_rtpoll_free(u->rtpoll);
2025
2026 if (u->pstream) {
2027 pa_pstream_unlink(u->pstream);
2028 pa_pstream_unref(u->pstream);
2029 }
2030
2031 if (u->pdispatch)
2032 pa_pdispatch_unref(u->pdispatch);
2033
2034 if (u->client)
2035 pa_socket_client_unref(u->client);
2036
2037 if (u->auth_cookie)
2038 pa_auth_cookie_unref(u->auth_cookie);
2039
2040 if (u->smoother)
2041 pa_smoother_free(u->smoother);
2042
2043 if (u->time_event)
2044 u->core->mainloop->time_free(u->time_event);
2045
2046 #ifndef TUNNEL_SINK
2047 if (u->mcalign)
2048 pa_mcalign_free(u->mcalign);
2049 #endif
2050
2051 #ifdef TUNNEL_SINK
2052 pa_xfree(u->sink_name);
2053 #else
2054 pa_xfree(u->source_name);
2055 #endif
2056 pa_xfree(u->server_name);
2057
2058 pa_xfree(u->device_description);
2059 pa_xfree(u->server_fqdn);
2060 pa_xfree(u->user_name);
2061
2062 pa_xfree(u);
2063 }