]> code.delx.au - pulseaudio/blob - src/modules/module-tunnel.c
delx.au / code / pulseaudio / blob
summary | shortlog | log | commit | commitdiff | tree
history | raw | HEAD
Merge remote branch 'origin/master-tx'
[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 = 0, 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 = FALSE;
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 pa_assert(u);
1349
1350 t = pa_tagstruct_new(NULL, 0);
1351 pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE);
1352 pa_tagstruct_putu32(t, u->ctag++);
1353 pa_tagstruct_putu32(t, PA_SUBSCRIPTION_MASK_SERVER|
1354 #ifdef TUNNEL_SINK
1355 PA_SUBSCRIPTION_MASK_SINK_INPUT|PA_SUBSCRIPTION_MASK_SINK
1356 #else
1357 PA_SUBSCRIPTION_MASK_SOURCE
1358 #endif
1359 );
1360
1361 pa_pstream_send_tagstruct(u->pstream, t);
1362 }
1363
1364 /* Called from main context */
1365 static void create_stream_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1366 struct userdata *u = userdata;
1367 #ifdef TUNNEL_SINK
1368 uint32_t bytes;
1369 #endif
1370
1371 pa_assert(pd);
1372 pa_assert(u);
1373 pa_assert(u->pdispatch == pd);
1374
1375 if (command != PA_COMMAND_REPLY) {
1376 if (command == PA_COMMAND_ERROR)
1377 pa_log("Failed to create stream.");
1378 else
1379 pa_log("Protocol error.");
1380 goto fail;
1381 }
1382
1383 if (pa_tagstruct_getu32(t, &u->channel) < 0 ||
1384 pa_tagstruct_getu32(t, &u->device_index) < 0
1385 #ifdef TUNNEL_SINK
1386 || pa_tagstruct_getu32(t, &bytes) < 0
1387 #endif
1388 )
1389 goto parse_error;
1390
1391 if (u->version >= 9) {
1392 #ifdef TUNNEL_SINK
1393 if (pa_tagstruct_getu32(t, &u->maxlength) < 0 ||
1394 pa_tagstruct_getu32(t, &u->tlength) < 0 ||
1395 pa_tagstruct_getu32(t, &u->prebuf) < 0 ||
1396 pa_tagstruct_getu32(t, &u->minreq) < 0)
1397 goto parse_error;
1398 #else
1399 if (pa_tagstruct_getu32(t, &u->maxlength) < 0 ||
1400 pa_tagstruct_getu32(t, &u->fragsize) < 0)
1401 goto parse_error;
1402 #endif
1403 }
1404
1405 if (u->version >= 12) {
1406 pa_sample_spec ss;
1407 pa_channel_map cm;
1408 uint32_t device_index;
1409 const char *dn;
1410 pa_bool_t suspended;
1411
1412 if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1413 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1414 pa_tagstruct_getu32(t, &device_index) < 0 ||
1415 pa_tagstruct_gets(t, &dn) < 0 ||
1416 pa_tagstruct_get_boolean(t, &suspended) < 0)
1417 goto parse_error;
1418
1419 #ifdef TUNNEL_SINK
1420 pa_xfree(u->sink_name);
1421 u->sink_name = pa_xstrdup(dn);
1422 #else
1423 pa_xfree(u->source_name);
1424 u->source_name = pa_xstrdup(dn);
1425 #endif
1426 }
1427
1428 if (u->version >= 13) {
1429 pa_usec_t usec;
1430
1431 if (pa_tagstruct_get_usec(t, &usec) < 0)
1432 goto parse_error;
1433
1434 /* #ifdef TUNNEL_SINK */
1435 /* pa_sink_set_latency_range(u->sink, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1436 /* #else */
1437 /* pa_source_set_latency_range(u->source, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1438 /* #endif */
1439 }
1440
1441 if (!pa_tagstruct_eof(t))
1442 goto parse_error;
1443
1444 start_subscribe(u);
1445 request_info(u);
1446
1447 pa_assert(!u->time_event);
1448 u->time_event = pa_core_rttime_new(u->core, pa_rtclock_now() + LATENCY_INTERVAL, timeout_callback, u);
1449
1450 request_latency(u);
1451
1452 pa_log_debug("Stream created.");
1453
1454 #ifdef TUNNEL_SINK
1455 pa_asyncmsgq_post(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REQUEST, NULL, bytes, NULL, NULL);
1456 #endif
1457
1458 return;
1459
1460 parse_error:
1461 pa_log("Invalid reply. (Create stream)");
1462
1463 fail:
1464 pa_module_unload_request(u->module, TRUE);
1465
1466 }
1467
1468 /* Called from main context */
1469 static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1470 struct userdata *u = userdata;
1471 pa_tagstruct *reply;
1472 char name[256], un[128], hn[128];
1473 #ifdef TUNNEL_SINK
1474 pa_cvolume volume;
1475 #endif
1476
1477 pa_assert(pd);
1478 pa_assert(u);
1479 pa_assert(u->pdispatch == pd);
1480
1481 if (command != PA_COMMAND_REPLY ||
1482 pa_tagstruct_getu32(t, &u->version) < 0 ||
1483 !pa_tagstruct_eof(t)) {
1484
1485 if (command == PA_COMMAND_ERROR)
1486 pa_log("Failed to authenticate");
1487 else
1488 pa_log("Protocol error.");
1489
1490 goto fail;
1491 }
1492
1493 /* Minimum supported protocol version */
1494 if (u->version < 8) {
1495 pa_log("Incompatible protocol version");
1496 goto fail;
1497 }
1498
1499 /* Starting with protocol version 13 the MSB of the version tag
1500 reflects if shm is enabled for this connection or not. We don't
1501 support SHM here at all, so we just ignore this. */
1502
1503 if (u->version >= 13)
1504 u->version &= 0x7FFFFFFFU;
1505
1506 pa_log_debug("Protocol version: remote %u, local %u", u->version, PA_PROTOCOL_VERSION);
1507
1508 #ifdef TUNNEL_SINK
1509 pa_proplist_setf(u->sink->proplist, "tunnel.remote_version", "%u", u->version);
1510 pa_sink_update_proplist(u->sink, 0, NULL);
1511
1512 pa_snprintf(name, sizeof(name), "%s for %s@%s",
1513 u->sink_name,
1514 pa_get_user_name(un, sizeof(un)),
1515 pa_get_host_name(hn, sizeof(hn)));
1516 #else
1517 pa_proplist_setf(u->source->proplist, "tunnel.remote_version", "%u", u->version);
1518 pa_source_update_proplist(u->source, 0, NULL);
1519
1520 pa_snprintf(name, sizeof(name), "%s for %s@%s",
1521 u->source_name,
1522 pa_get_user_name(un, sizeof(un)),
1523 pa_get_host_name(hn, sizeof(hn)));
1524 #endif
1525
1526 reply = pa_tagstruct_new(NULL, 0);
1527 pa_tagstruct_putu32(reply, PA_COMMAND_SET_CLIENT_NAME);
1528 pa_tagstruct_putu32(reply, u->ctag++);
1529
1530 if (u->version >= 13) {
1531 pa_proplist *pl;
1532 pl = pa_proplist_new();
1533 pa_proplist_sets(pl, PA_PROP_APPLICATION_ID, "org.PulseAudio.PulseAudio");
1534 pa_proplist_sets(pl, PA_PROP_APPLICATION_VERSION, PACKAGE_VERSION);
1535 pa_init_proplist(pl);
1536 pa_tagstruct_put_proplist(reply, pl);
1537 pa_proplist_free(pl);
1538 } else
1539 pa_tagstruct_puts(reply, "PulseAudio");
1540
1541 pa_pstream_send_tagstruct(u->pstream, reply);
1542 /* We ignore the server's reply here */
1543
1544 reply = pa_tagstruct_new(NULL, 0);
1545
1546 if (u->version < 13)
1547 /* Only for older PA versions we need to fill in the maxlength */
1548 u->maxlength = 4*1024*1024;
1549
1550 #ifdef TUNNEL_SINK
1551 u->tlength = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_TLENGTH_MSEC, &u->sink->sample_spec);
1552 u->minreq = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_MINREQ_MSEC, &u->sink->sample_spec);
1553 u->prebuf = u->tlength;
1554 #else
1555 u->fragsize = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_FRAGSIZE_MSEC, &u->source->sample_spec);
1556 #endif
1557
1558 #ifdef TUNNEL_SINK
1559 pa_tagstruct_putu32(reply, PA_COMMAND_CREATE_PLAYBACK_STREAM);
1560 pa_tagstruct_putu32(reply, tag = u->ctag++);
1561
1562 if (u->version < 13)
1563 pa_tagstruct_puts(reply, name);
1564
1565 pa_tagstruct_put_sample_spec(reply, &u->sink->sample_spec);
1566 pa_tagstruct_put_channel_map(reply, &u->sink->channel_map);
1567 pa_tagstruct_putu32(reply, PA_INVALID_INDEX);
1568 pa_tagstruct_puts(reply, u->sink_name);
1569 pa_tagstruct_putu32(reply, u->maxlength);
1570 pa_tagstruct_put_boolean(reply, !PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)));
1571 pa_tagstruct_putu32(reply, u->tlength);
1572 pa_tagstruct_putu32(reply, u->prebuf);
1573 pa_tagstruct_putu32(reply, u->minreq);
1574 pa_tagstruct_putu32(reply, 0);
1575 pa_cvolume_reset(&volume, u->sink->sample_spec.channels);
1576 pa_tagstruct_put_cvolume(reply, &volume);
1577 #else
1578 pa_tagstruct_putu32(reply, PA_COMMAND_CREATE_RECORD_STREAM);
1579 pa_tagstruct_putu32(reply, tag = u->ctag++);
1580
1581 if (u->version < 13)
1582 pa_tagstruct_puts(reply, name);
1583
1584 pa_tagstruct_put_sample_spec(reply, &u->source->sample_spec);
1585 pa_tagstruct_put_channel_map(reply, &u->source->channel_map);
1586 pa_tagstruct_putu32(reply, PA_INVALID_INDEX);
1587 pa_tagstruct_puts(reply, u->source_name);
1588 pa_tagstruct_putu32(reply, u->maxlength);
1589 pa_tagstruct_put_boolean(reply, !PA_SOURCE_IS_OPENED(pa_source_get_state(u->source)));
1590 pa_tagstruct_putu32(reply, u->fragsize);
1591 #endif
1592
1593 if (u->version >= 12) {
1594 pa_tagstruct_put_boolean(reply, FALSE); /* no_remap */
1595 pa_tagstruct_put_boolean(reply, FALSE); /* no_remix */
1596 pa_tagstruct_put_boolean(reply, FALSE); /* fix_format */
1597 pa_tagstruct_put_boolean(reply, FALSE); /* fix_rate */
1598 pa_tagstruct_put_boolean(reply, FALSE); /* fix_channels */
1599 pa_tagstruct_put_boolean(reply, TRUE); /* no_move */
1600 pa_tagstruct_put_boolean(reply, FALSE); /* variable_rate */
1601 }
1602
1603 if (u->version >= 13) {
1604 pa_proplist *pl;
1605
1606 pa_tagstruct_put_boolean(reply, FALSE); /* start muted/peak detect*/
1607 pa_tagstruct_put_boolean(reply, TRUE); /* adjust_latency */
1608
1609 pl = pa_proplist_new();
1610 pa_proplist_sets(pl, PA_PROP_MEDIA_NAME, name);
1611 pa_proplist_sets(pl, PA_PROP_MEDIA_ROLE, "abstract");
1612 pa_tagstruct_put_proplist(reply, pl);
1613 pa_proplist_free(pl);
1614
1615 #ifndef TUNNEL_SINK
1616 pa_tagstruct_putu32(reply, PA_INVALID_INDEX); /* direct on input */
1617 #endif
1618 }
1619
1620 if (u->version >= 14) {
1621 #ifdef TUNNEL_SINK
1622 pa_tagstruct_put_boolean(reply, FALSE); /* volume_set */
1623 #endif
1624 pa_tagstruct_put_boolean(reply, TRUE); /* early rquests */
1625 }
1626
1627 if (u->version >= 15) {
1628 #ifdef TUNNEL_SINK
1629 pa_tagstruct_put_boolean(reply, FALSE); /* muted_set */
1630 #endif
1631 pa_tagstruct_put_boolean(reply, FALSE); /* don't inhibit auto suspend */
1632 pa_tagstruct_put_boolean(reply, FALSE); /* fail on suspend */
1633 }
1634
1635 pa_pstream_send_tagstruct(u->pstream, reply);
1636 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, create_stream_callback, u, NULL);
1637
1638 pa_log_debug("Connection authenticated, creating stream ...");
1639
1640 return;
1641
1642 fail:
1643 pa_module_unload_request(u->module, TRUE);
1644 }
1645
1646 /* Called from main context */
1647 static void pstream_die_callback(pa_pstream *p, void *userdata) {
1648 struct userdata *u = userdata;
1649
1650 pa_assert(p);
1651 pa_assert(u);
1652
1653 pa_log_warn("Stream died.");
1654 pa_module_unload_request(u->module, TRUE);
1655 }
1656
1657 /* Called from main context */
1658 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
1659 struct userdata *u = userdata;
1660
1661 pa_assert(p);
1662 pa_assert(packet);
1663 pa_assert(u);
1664
1665 if (pa_pdispatch_run(u->pdispatch, packet, creds, u) < 0) {
1666 pa_log("Invalid packet");
1667 pa_module_unload_request(u->module, TRUE);
1668 return;
1669 }
1670 }
1671
1672 #ifndef TUNNEL_SINK
1673 /* Called from main context */
1674 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) {
1675 struct userdata *u = userdata;
1676
1677 pa_assert(p);
1678 pa_assert(chunk);
1679 pa_assert(u);
1680
1681 if (channel != u->channel) {
1682 pa_log("Received memory block on bad channel.");
1683 pa_module_unload_request(u->module, TRUE);
1684 return;
1685 }
1686
1687 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_POST, PA_UINT_TO_PTR(seek), offset, chunk);
1688
1689 u->counter_delta += (int64_t) chunk->length;
1690 }
1691 #endif
1692
1693 /* Called from main context */
1694 static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) {
1695 struct userdata *u = userdata;
1696 pa_tagstruct *t;
1697 uint32_t tag;
1698
1699 pa_assert(sc);
1700 pa_assert(u);
1701 pa_assert(u->client == sc);
1702
1703 pa_socket_client_unref(u->client);
1704 u->client = NULL;
1705
1706 if (!io) {
1707 pa_log("Connection failed: %s", pa_cstrerror(errno));
1708 pa_module_unload_request(u->module, TRUE);
1709 return;
1710 }
1711
1712 u->pstream = pa_pstream_new(u->core->mainloop, io, u->core->mempool);
1713 u->pdispatch = pa_pdispatch_new(u->core->mainloop, TRUE, command_table, PA_COMMAND_MAX);
1714
1715 pa_pstream_set_die_callback(u->pstream, pstream_die_callback, u);
1716 pa_pstream_set_recieve_packet_callback(u->pstream, pstream_packet_callback, u);
1717 #ifndef TUNNEL_SINK
1718 pa_pstream_set_recieve_memblock_callback(u->pstream, pstream_memblock_callback, u);
1719 #endif
1720
1721 t = pa_tagstruct_new(NULL, 0);
1722 pa_tagstruct_putu32(t, PA_COMMAND_AUTH);
1723 pa_tagstruct_putu32(t, tag = u->ctag++);
1724 pa_tagstruct_putu32(t, PA_PROTOCOL_VERSION);
1725
1726 pa_tagstruct_put_arbitrary(t, pa_auth_cookie_read(u->auth_cookie, PA_NATIVE_COOKIE_LENGTH), PA_NATIVE_COOKIE_LENGTH);
1727
1728 #ifdef HAVE_CREDS
1729 {
1730 pa_creds ucred;
1731
1732 if (pa_iochannel_creds_supported(io))
1733 pa_iochannel_creds_enable(io);
1734
1735 ucred.uid = getuid();
1736 ucred.gid = getgid();
1737
1738 pa_pstream_send_tagstruct_with_creds(u->pstream, t, &ucred);
1739 }
1740 #else
1741 pa_pstream_send_tagstruct(u->pstream, t);
1742 #endif
1743
1744 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, u, NULL);
1745
1746 pa_log_debug("Connection established, authenticating ...");
1747 }
1748
1749 #ifdef TUNNEL_SINK
1750
1751 /* Called from main context */
1752 static void sink_set_volume(pa_sink *sink) {
1753 struct userdata *u;
1754 pa_tagstruct *t;
1755
1756 pa_assert(sink);
1757 u = sink->userdata;
1758 pa_assert(u);
1759
1760 t = pa_tagstruct_new(NULL, 0);
1761 pa_tagstruct_putu32(t, PA_COMMAND_SET_SINK_INPUT_VOLUME);
1762 pa_tagstruct_putu32(t, u->ctag++);
1763 pa_tagstruct_putu32(t, u->device_index);
1764 pa_tagstruct_put_cvolume(t, &sink->real_volume);
1765 pa_pstream_send_tagstruct(u->pstream, t);
1766 }
1767
1768 /* Called from main context */
1769 static void sink_set_mute(pa_sink *sink) {
1770 struct userdata *u;
1771 pa_tagstruct *t;
1772
1773 pa_assert(sink);
1774 u = sink->userdata;
1775 pa_assert(u);
1776
1777 if (u->version < 11)
1778 return;
1779
1780 t = pa_tagstruct_new(NULL, 0);
1781 pa_tagstruct_putu32(t, PA_COMMAND_SET_SINK_INPUT_MUTE);
1782 pa_tagstruct_putu32(t, u->ctag++);
1783 pa_tagstruct_putu32(t, u->device_index);
1784 pa_tagstruct_put_boolean(t, !!sink->muted);
1785 pa_pstream_send_tagstruct(u->pstream, t);
1786 }
1787
1788 #endif
1789
1790 int pa__init(pa_module*m) {
1791 pa_modargs *ma = NULL;
1792 struct userdata *u = NULL;
1793 pa_sample_spec ss;
1794 pa_channel_map map;
1795 char *dn = NULL;
1796 #ifdef TUNNEL_SINK
1797 pa_sink_new_data data;
1798 #else
1799 pa_source_new_data data;
1800 #endif
1801
1802 pa_assert(m);
1803
1804 if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
1805 pa_log("Failed to parse module arguments");
1806 goto fail;
1807 }
1808
1809 m->userdata = u = pa_xnew0(struct userdata, 1);
1810 u->core = m->core;
1811 u->module = m;
1812 u->client = NULL;
1813 u->pdispatch = NULL;
1814 u->pstream = NULL;
1815 u->server_name = NULL;
1816 #ifdef TUNNEL_SINK
1817 u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));;
1818 u->sink = NULL;
1819 u->requested_bytes = 0;
1820 #else
1821 u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));;
1822 u->source = NULL;
1823 #endif
1824 u->smoother = pa_smoother_new(
1825 PA_USEC_PER_SEC,
1826 PA_USEC_PER_SEC*2,
1827 TRUE,
1828 TRUE,
1829 10,
1830 pa_rtclock_now(),
1831 FALSE);
1832 u->ctag = 1;
1833 u->device_index = u->channel = PA_INVALID_INDEX;
1834 u->time_event = NULL;
1835 u->ignore_latency_before = 0;
1836 u->transport_usec = u->thread_transport_usec = 0;
1837 u->remote_suspended = u->remote_corked = FALSE;
1838 u->counter = u->counter_delta = 0;
1839
1840 u->rtpoll = pa_rtpoll_new();
1841 pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
1842
1843 if (!(u->auth_cookie = pa_auth_cookie_get(u->core, pa_modargs_get_value(ma, "cookie", PA_NATIVE_COOKIE_FILE), PA_NATIVE_COOKIE_LENGTH)))
1844 goto fail;
1845
1846 if (!(u->server_name = pa_xstrdup(pa_modargs_get_value(ma, "server", NULL)))) {
1847 pa_log("No server specified.");
1848 goto fail;
1849 }
1850
1851 ss = m->core->default_sample_spec;
1852 map = m->core->default_channel_map;
1853 if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
1854 pa_log("Invalid sample format specification");
1855 goto fail;
1856 }
1857
1858 if (!(u->client = pa_socket_client_new_string(m->core->mainloop, TRUE, u->server_name, PA_NATIVE_DEFAULT_PORT))) {
1859 pa_log("Failed to connect to server '%s'", u->server_name);
1860 goto fail;
1861 }
1862
1863 pa_socket_client_set_callback(u->client, on_connection, u);
1864
1865 #ifdef TUNNEL_SINK
1866
1867 if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL))))
1868 dn = pa_sprintf_malloc("tunnel.%s", u->server_name);
1869
1870 pa_sink_new_data_init(&data);
1871 data.driver = __FILE__;
1872 data.module = m;
1873 data.namereg_fail = TRUE;
1874 pa_sink_new_data_set_name(&data, dn);
1875 pa_sink_new_data_set_sample_spec(&data, &ss);
1876 pa_sink_new_data_set_channel_map(&data, &map);
1877 pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->sink_name), u->sink_name ? " on " : "", u->server_name);
1878 pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
1879 if (u->sink_name)
1880 pa_proplist_sets(data.proplist, "tunnel.remote.sink", u->sink_name);
1881
1882 if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
1883 pa_log("Invalid properties");
1884 pa_sink_new_data_done(&data);
1885 goto fail;
1886 }
1887
1888 u->sink = pa_sink_new(m->core, &data, PA_SINK_NETWORK|PA_SINK_LATENCY|PA_SINK_HW_VOLUME_CTRL|PA_SINK_HW_MUTE_CTRL);
1889 pa_sink_new_data_done(&data);
1890
1891 if (!u->sink) {
1892 pa_log("Failed to create sink.");
1893 goto fail;
1894 }
1895
1896 u->sink->parent.process_msg = sink_process_msg;
1897 u->sink->userdata = u;
1898 u->sink->set_state = sink_set_state;
1899 u->sink->set_volume = sink_set_volume;
1900 u->sink->set_mute = sink_set_mute;
1901
1902 u->sink->refresh_volume = u->sink->refresh_muted = FALSE;
1903
1904 /* pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
1905
1906 pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
1907 pa_sink_set_rtpoll(u->sink, u->rtpoll);
1908
1909 #else
1910
1911 if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL))))
1912 dn = pa_sprintf_malloc("tunnel.%s", u->server_name);
1913
1914 pa_source_new_data_init(&data);
1915 data.driver = __FILE__;
1916 data.module = m;
1917 data.namereg_fail = TRUE;
1918 pa_source_new_data_set_name(&data, dn);
1919 pa_source_new_data_set_sample_spec(&data, &ss);
1920 pa_source_new_data_set_channel_map(&data, &map);
1921 pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->source_name), u->source_name ? " on " : "", u->server_name);
1922 pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
1923 if (u->source_name)
1924 pa_proplist_sets(data.proplist, "tunnel.remote.source", u->source_name);
1925
1926 if (pa_modargs_get_proplist(ma, "source_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
1927 pa_log("Invalid properties");
1928 pa_source_new_data_done(&data);
1929 goto fail;
1930 }
1931
1932 u->source = pa_source_new(m->core, &data, PA_SOURCE_NETWORK|PA_SOURCE_LATENCY);
1933 pa_source_new_data_done(&data);
1934
1935 if (!u->source) {
1936 pa_log("Failed to create source.");
1937 goto fail;
1938 }
1939
1940 u->source->parent.process_msg = source_process_msg;
1941 u->source->set_state = source_set_state;
1942 u->source->userdata = u;
1943
1944 /* pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
1945
1946 pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
1947 pa_source_set_rtpoll(u->source, u->rtpoll);
1948
1949 u->mcalign = pa_mcalign_new(pa_frame_size(&u->source->sample_spec));
1950 #endif
1951
1952 pa_xfree(dn);
1953
1954 u->time_event = NULL;
1955
1956 u->maxlength = (uint32_t) -1;
1957 #ifdef TUNNEL_SINK
1958 u->tlength = u->minreq = u->prebuf = (uint32_t) -1;
1959 #else
1960 u->fragsize = (uint32_t) -1;
1961 #endif
1962
1963 if (!(u->thread = pa_thread_new(thread_func, u))) {
1964 pa_log("Failed to create thread.");
1965 goto fail;
1966 }
1967
1968 #ifdef TUNNEL_SINK
1969 pa_sink_put(u->sink);
1970 #else
1971 pa_source_put(u->source);
1972 #endif
1973
1974 pa_modargs_free(ma);
1975
1976 return 0;
1977
1978 fail:
1979 pa__done(m);
1980
1981 if (ma)
1982 pa_modargs_free(ma);
1983
1984 pa_xfree(dn);
1985
1986 return -1;
1987 }
1988
1989 void pa__done(pa_module*m) {
1990 struct userdata* u;
1991
1992 pa_assert(m);
1993
1994 if (!(u = m->userdata))
1995 return;
1996
1997 #ifdef TUNNEL_SINK
1998 if (u->sink)
1999 pa_sink_unlink(u->sink);
2000 #else
2001 if (u->source)
2002 pa_source_unlink(u->source);
2003 #endif
2004
2005 if (u->thread) {
2006 pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
2007 pa_thread_free(u->thread);
2008 }
2009
2010 pa_thread_mq_done(&u->thread_mq);
2011
2012 #ifdef TUNNEL_SINK
2013 if (u->sink)
2014 pa_sink_unref(u->sink);
2015 #else
2016 if (u->source)
2017 pa_source_unref(u->source);
2018 #endif
2019
2020 if (u->rtpoll)
2021 pa_rtpoll_free(u->rtpoll);
2022
2023 if (u->pstream) {
2024 pa_pstream_unlink(u->pstream);
2025 pa_pstream_unref(u->pstream);
2026 }
2027
2028 if (u->pdispatch)
2029 pa_pdispatch_unref(u->pdispatch);
2030
2031 if (u->client)
2032 pa_socket_client_unref(u->client);
2033
2034 if (u->auth_cookie)
2035 pa_auth_cookie_unref(u->auth_cookie);
2036
2037 if (u->smoother)
2038 pa_smoother_free(u->smoother);
2039
2040 if (u->time_event)
2041 u->core->mainloop->time_free(u->time_event);
2042
2043 #ifndef TUNNEL_SINK
2044 if (u->mcalign)
2045 pa_mcalign_free(u->mcalign);
2046 #endif
2047
2048 #ifdef TUNNEL_SINK
2049 pa_xfree(u->sink_name);
2050 #else
2051 pa_xfree(u->source_name);
2052 #endif
2053 pa_xfree(u->server_name);
2054
2055 pa_xfree(u->device_description);
2056 pa_xfree(u->server_fqdn);
2057 pa_xfree(u->user_name);
2058
2059 pa_xfree(u);
2060 }