2 This file is part of PulseAudio.
4 Copyright 2004-2006 Lennart Poettering
5 Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
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.
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.
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
30 #include <sys/types.h>
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>
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>
61 #include "module-tunnel-sink-symdef.h"
63 #include "module-tunnel-source-symdef.h"
67 PA_MODULE_DESCRIPTION("Tunnel module for sinks");
69 "sink_name=<name for the local sink> "
70 "sink_properties=<properties for the local sink> "
72 "sink=<remote sink name> "
74 "format=<sample format> "
75 "channels=<number of channels> "
77 "channel_map=<channel map>");
79 PA_MODULE_DESCRIPTION("Tunnel module for sources");
81 "source_name=<name for the local source> "
82 "source_properties=<properties for the local source> "
84 "source=<remote source name> "
86 "format=<sample format> "
87 "channels=<number of channels> "
89 "channel_map=<channel map>");
92 PA_MODULE_AUTHOR("Lennart Poettering");
93 PA_MODULE_VERSION(PACKAGE_VERSION
);
94 PA_MODULE_LOAD_ONCE(FALSE
);
96 static const char* const valid_modargs
[] = {
115 #define DEFAULT_TIMEOUT 5
117 #define LATENCY_INTERVAL (10*PA_USEC_PER_SEC)
119 #define MIN_NETWORK_LATENCY_USEC (8*PA_USEC_PER_MSEC)
124 SINK_MESSAGE_REQUEST
= PA_SINK_MESSAGE_MAX
,
125 SINK_MESSAGE_REMOTE_SUSPEND
,
126 SINK_MESSAGE_UPDATE_LATENCY
,
130 #define DEFAULT_TLENGTH_MSEC 150
131 #define DEFAULT_MINREQ_MSEC 25
136 SOURCE_MESSAGE_POST
= PA_SOURCE_MESSAGE_MAX
,
137 SOURCE_MESSAGE_REMOTE_SUSPEND
,
138 SOURCE_MESSAGE_UPDATE_LATENCY
141 #define DEFAULT_FRAGSIZE_MSEC 25
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
);
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
);
157 static const pa_pdispatch_cb_t command_table
[PA_COMMAND_MAX
] = {
159 [PA_COMMAND_REQUEST
] = command_request
,
160 [PA_COMMAND_STARTED
] = command_started
,
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
182 pa_thread_mq thread_mq
;
186 pa_socket_client
*client
;
188 pa_pdispatch
*pdispatch
;
194 size_t requested_bytes
;
201 pa_auth_cookie
*auth_cookie
;
205 uint32_t device_index
;
208 int64_t counter
, counter_delta
;
210 pa_bool_t remote_corked
:1;
211 pa_bool_t remote_suspended
:1;
213 pa_usec_t transport_usec
; /* maintained in the main thread */
214 pa_usec_t thread_transport_usec
; /* maintained in the IO thread */
216 uint32_t ignore_latency_before
;
218 pa_time_event
*time_event
;
220 pa_smoother
*smoother
;
222 char *device_description
;
236 static void request_latency(struct userdata
*u
);
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.");
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
;
250 pa_assert(u
->pdispatch
== pd
);
252 pa_log_warn("Stream killed");
253 pa_module_unload_request(u
->module
, TRUE
);
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
;
263 pa_assert(u
->pdispatch
== pd
);
265 pa_log_info("Server signalled buffer overrun/underrun.");
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
;
278 pa_assert(u
->pdispatch
== pd
);
280 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
281 pa_tagstruct_get_boolean(t
, &suspended
) < 0 ||
282 !pa_tagstruct_eof(t
)) {
284 pa_log("Invalid packet.");
285 pa_module_unload_request(u
->module
, TRUE
);
289 pa_log_debug("Server reports device suspend.");
292 pa_asyncmsgq_send(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
294 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
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
;
310 pa_assert(u
->pdispatch
== pd
);
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) {
317 pa_log_error("Invalid packet.");
318 pa_module_unload_request(u
->module
, TRUE
);
322 pa_log_debug("Server reports a stream move.");
325 pa_asyncmsgq_send(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
327 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
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
;
341 pa_assert(u
->pdispatch
== pd
);
343 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
344 pa_tagstruct_getu32(t
, &maxlength
) < 0) {
346 pa_log_error("Invalid packet.");
347 pa_module_unload_request(u
->module
, TRUE
);
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) {
355 pa_log_error("Invalid packet.");
356 pa_module_unload_request(u
->module
, TRUE
);
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) {
365 pa_log_error("Invalid packet.");
366 pa_module_unload_request(u
->module
, TRUE
);
372 pa_log_debug("Server reports buffer attrs changed. tlength now at %lu, before %lu.", (unsigned long) tlength
, (unsigned long) u
->tlength
);
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
;
387 pa_assert(u
->pdispatch
== pd
);
389 pa_log_debug("Server reports playback started.");
395 /* Called from IO thread context */
396 static void check_smoother_status(struct userdata
*u
, pa_bool_t past
) {
401 x
= pa_rtclock_now();
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 */
408 x
-= u
->thread_transport_usec
;
410 x
+= u
->thread_transport_usec
;
412 if (u
->remote_suspended
|| u
->remote_corked
)
413 pa_smoother_pause(u
->smoother
, x
);
415 pa_smoother_resume(u
->smoother
, x
, TRUE
);
418 /* Called from IO thread context */
419 static void stream_cork_within_thread(struct userdata
*u
, pa_bool_t cork
) {
422 if (u
->remote_corked
== cork
)
425 u
->remote_corked
= cork
;
426 check_smoother_status(u
, FALSE
);
429 /* Called from main context */
430 static void stream_cork(struct userdata
*u
, pa_bool_t cork
) {
437 t
= pa_tagstruct_new(NULL
, 0);
439 pa_tagstruct_putu32(t
, PA_COMMAND_CORK_PLAYBACK_STREAM
);
441 pa_tagstruct_putu32(t
, PA_COMMAND_CORK_RECORD_STREAM
);
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
);
451 /* Called from IO thread context */
452 static void stream_suspend_within_thread(struct userdata
*u
, pa_bool_t suspend
) {
455 if (u
->remote_suspended
== suspend
)
458 u
->remote_suspended
= suspend
;
459 check_smoother_status(u
, TRUE
);
464 /* Called from IO thread context */
465 static void send_data(struct userdata
*u
) {
468 while (u
->requested_bytes
> 0) {
469 pa_memchunk memchunk
;
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
);
475 u
->requested_bytes
-= memchunk
.length
;
477 u
->counter
+= (int64_t) memchunk
.length
;
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
;
487 case PA_SINK_MESSAGE_SET_STATE
: {
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) {
493 stream_cork_within_thread(u
, u
->sink
->state
== PA_SINK_SUSPENDED
);
495 if (PA_SINK_IS_OPENED(u
->sink
->state
))
502 case PA_SINK_MESSAGE_GET_LATENCY
: {
503 pa_usec_t yl
, yr
, *usec
= data
;
505 yl
= pa_bytes_to_usec((uint64_t) u
->counter
, &u
->sink
->sample_spec
);
506 yr
= pa_smoother_get(u
->smoother
, pa_rtclock_now());
508 *usec
= yl
> yr
? yl
- yr
: 0;
512 case SINK_MESSAGE_REQUEST
:
514 pa_assert(offset
> 0);
515 u
->requested_bytes
+= (size_t) offset
;
517 if (PA_SINK_IS_OPENED(u
->sink
->thread_info
.state
))
523 case SINK_MESSAGE_REMOTE_SUSPEND
:
525 stream_suspend_within_thread(u
, !!PA_PTR_TO_UINT(data
));
529 case SINK_MESSAGE_UPDATE_LATENCY
: {
532 y
= pa_bytes_to_usec((uint64_t) u
->counter
, &u
->sink
->sample_spec
);
534 if (y
> (pa_usec_t
) offset
)
535 y
-= (pa_usec_t
) offset
;
539 pa_smoother_put(u
->smoother
, pa_rtclock_now(), y
);
541 /* We can access this freely here, since the main thread is waiting for us */
542 u
->thread_transport_usec
= u
->transport_usec
;
547 case SINK_MESSAGE_POST
:
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. */
554 pa_pstream_send_memblock(u
->pstream
, u
->channel
, 0, PA_SEEK_RELATIVE
, chunk
);
556 u
->counter_delta
+= (int64_t) chunk
->length
;
561 return pa_sink_process_msg(o
, code
, data
, offset
, chunk
);
564 /* Called from main context */
565 static int sink_set_state(pa_sink
*s
, pa_sink_state_t state
) {
567 pa_sink_assert_ref(s
);
570 switch ((pa_sink_state_t
) state
) {
572 case PA_SINK_SUSPENDED
:
573 pa_assert(PA_SINK_IS_OPENED(s
->state
));
574 stream_cork(u
, TRUE
);
578 case PA_SINK_RUNNING
:
579 if (s
->state
== PA_SINK_SUSPENDED
)
580 stream_cork(u
, FALSE
);
583 case PA_SINK_UNLINKED
:
585 case PA_SINK_INVALID_STATE
:
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
;
600 case PA_SOURCE_MESSAGE_SET_STATE
: {
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
);
609 case PA_SOURCE_MESSAGE_GET_LATENCY
: {
610 pa_usec_t yr
, yl
, *usec
= data
;
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());
615 *usec
= yr
> yl
? yr
- yl
: 0;
619 case SOURCE_MESSAGE_POST
: {
622 pa_mcalign_push(u
->mcalign
, chunk
);
624 while (pa_mcalign_pop(u
->mcalign
, &c
) >= 0) {
626 if (PA_SOURCE_IS_OPENED(u
->source
->thread_info
.state
))
627 pa_source_post(u
->source
, &c
);
629 pa_memblock_unref(c
.memblock
);
631 u
->counter
+= (int64_t) c
.length
;
637 case SOURCE_MESSAGE_REMOTE_SUSPEND
:
639 stream_suspend_within_thread(u
, !!PA_PTR_TO_UINT(data
));
642 case SOURCE_MESSAGE_UPDATE_LATENCY
: {
645 y
= pa_bytes_to_usec((uint64_t) u
->counter
, &u
->source
->sample_spec
);
646 y
+= (pa_usec_t
) offset
;
648 pa_smoother_put(u
->smoother
, pa_rtclock_now(), y
);
650 /* We can access this freely here, since the main thread is waiting for us */
651 u
->thread_transport_usec
= u
->transport_usec
;
657 return pa_source_process_msg(o
, code
, data
, offset
, chunk
);
660 /* Called from main context */
661 static int source_set_state(pa_source
*s
, pa_source_state_t state
) {
663 pa_source_assert_ref(s
);
666 switch ((pa_source_state_t
) state
) {
668 case PA_SOURCE_SUSPENDED
:
669 pa_assert(PA_SOURCE_IS_OPENED(s
->state
));
670 stream_cork(u
, TRUE
);
674 case PA_SOURCE_RUNNING
:
675 if (s
->state
== PA_SOURCE_SUSPENDED
)
676 stream_cork(u
, FALSE
);
679 case PA_SOURCE_UNLINKED
:
681 case PA_SINK_INVALID_STATE
:
690 static void thread_func(void *userdata
) {
691 struct userdata
*u
= userdata
;
695 pa_log_debug("Thread starting up");
697 pa_thread_mq_install(&u
->thread_mq
);
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);
708 if ((ret
= pa_rtpoll_run(u
->rtpoll
, TRUE
)) < 0)
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
);
722 pa_log_debug("Thread shutting down");
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
;
732 pa_assert(command
== PA_COMMAND_REQUEST
);
735 pa_assert(u
->pdispatch
== pd
);
737 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
738 pa_tagstruct_getu32(t
, &bytes
) < 0) {
739 pa_log("Invalid protocol reply");
743 if (channel
!= u
->channel
) {
744 pa_log("Received data for invalid channel");
748 pa_asyncmsgq_post(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REQUEST
, NULL
, bytes
, NULL
, NULL
);
752 pa_module_unload_request(u
->module
, TRUE
);
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
;
762 int64_t write_index
, read_index
;
763 struct timeval local
, remote
, now
;
770 if (command
!= PA_COMMAND_REPLY
) {
771 if (command
== PA_COMMAND_ERROR
)
772 pa_log("Failed to get latency.");
774 pa_log("Protocol error.");
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.");
790 if (u
->version
>= 13) {
791 uint64_t underrun_for
= 0, playing_for
= 0;
793 if (pa_tagstruct_getu64(t
, &underrun_for
) < 0 ||
794 pa_tagstruct_getu64(t
, &playing_for
) < 0) {
795 pa_log("Invalid reply.");
801 if (!pa_tagstruct_eof(t
)) {
802 pa_log("Invalid reply.");
806 if (tag
< u
->ignore_latency_before
) {
810 pa_gettimeofday(&now
);
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 */
816 u
->transport_usec
= pa_timeval_diff(&remote
, &local
);
818 u
->transport_usec
= pa_timeval_diff(&now
, &remote
);
821 u
->transport_usec
= pa_timeval_diff(&now
, &local
)/2;
823 /* First, take the device's delay */
825 delay
= (int64_t) sink_usec
;
826 ss
= &u
->sink
->sample_spec
;
828 delay
= (int64_t) source_usec
;
829 ss
= &u
->source
->sample_spec
;
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
);
836 delay
-= (int64_t) pa_bytes_to_usec((uint64_t) (read_index
-write_index
), ss
);
838 /* Our measurements are already out of date, hence correct by the *
839 * transport latency */
841 delay
-= (int64_t) u
->transport_usec
;
843 delay
+= (int64_t) u
->transport_usec
;
846 /* Now correct by what we have have read/written since we requested the update */
848 delay
+= (int64_t) pa_bytes_to_usec((uint64_t) u
->counter_delta
, ss
);
850 delay
-= (int64_t) pa_bytes_to_usec((uint64_t) u
->counter_delta
, ss
);
854 pa_asyncmsgq_send(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_UPDATE_LATENCY
, 0, delay
, NULL
);
856 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_UPDATE_LATENCY
, 0, delay
, NULL
);
863 pa_module_unload_request(u
->module
, TRUE
);
866 /* Called from main context */
867 static void request_latency(struct userdata
*u
) {
873 t
= pa_tagstruct_new(NULL
, 0);
875 pa_tagstruct_putu32(t
, PA_COMMAND_GET_PLAYBACK_LATENCY
);
877 pa_tagstruct_putu32(t
, PA_COMMAND_GET_RECORD_LATENCY
);
879 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
880 pa_tagstruct_putu32(t
, u
->channel
);
882 pa_tagstruct_put_timeval(t
, pa_gettimeofday(&now
));
884 pa_pstream_send_tagstruct(u
->pstream
, t
);
885 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, stream_get_latency_callback
, u
, NULL
);
887 u
->ignore_latency_before
= tag
;
888 u
->counter_delta
= 0;
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
;
901 pa_core_rttime_restart(u
->core
, e
, pa_rtclock_now() + LATENCY_INTERVAL
);
904 /* Called from main context */
905 static void update_description(struct userdata
*u
) {
907 char un
[128], hn
[128];
912 if (!u
->server_fqdn
|| !u
->user_name
|| !u
->device_description
)
915 d
= pa_sprintf_malloc("%s on %s@%s", u
->device_description
, u
->user_name
, u
->server_fqdn
);
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
);
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
);
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
)));
935 t
= pa_tagstruct_new(NULL
, 0);
937 pa_tagstruct_putu32(t
, PA_COMMAND_SET_PLAYBACK_STREAM_NAME
);
939 pa_tagstruct_putu32(t
, PA_COMMAND_SET_RECORD_STREAM_NAME
);
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
);
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
;
954 const char *server_name
, *server_version
, *user_name
, *host_name
, *default_sink_name
, *default_source_name
;
960 if (command
!= PA_COMMAND_REPLY
) {
961 if (command
== PA_COMMAND_ERROR
)
962 pa_log("Failed to get info.");
964 pa_log("Protocol error.");
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 && pa_tagstruct_get_channel_map(t
, &cm
) < 0)) {
978 pa_log("Parse failure");
982 if (!pa_tagstruct_eof(t
)) {
983 pa_log("Packet too long");
987 pa_xfree(u
->server_fqdn
);
988 u
->server_fqdn
= pa_xstrdup(host_name
);
990 pa_xfree(u
->user_name
);
991 u
->user_name
= pa_xstrdup(user_name
);
993 update_description(u
);
998 pa_module_unload_request(u
->module
, TRUE
);
1003 /* Called from main context */
1004 static void sink_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1005 struct userdata
*u
= userdata
;
1006 uint32_t idx
, owner_module
, monitor_source
, flags
;
1007 const char *name
, *description
, *monitor_source_name
, *driver
;
1018 pl
= pa_proplist_new();
1020 if (command
!= PA_COMMAND_REPLY
) {
1021 if (command
== PA_COMMAND_ERROR
)
1022 pa_log("Failed to get info.");
1024 pa_log("Protocol error.");
1028 if (pa_tagstruct_getu32(t
, &idx
) < 0 ||
1029 pa_tagstruct_gets(t
, &name
) < 0 ||
1030 pa_tagstruct_gets(t
, &description
) < 0 ||
1031 pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
1032 pa_tagstruct_get_channel_map(t
, &cm
) < 0 ||
1033 pa_tagstruct_getu32(t
, &owner_module
) < 0 ||
1034 pa_tagstruct_get_cvolume(t
, &volume
) < 0 ||
1035 pa_tagstruct_get_boolean(t
, &mute
) < 0 ||
1036 pa_tagstruct_getu32(t
, &monitor_source
) < 0 ||
1037 pa_tagstruct_gets(t
, &monitor_source_name
) < 0 ||
1038 pa_tagstruct_get_usec(t
, &latency
) < 0 ||
1039 pa_tagstruct_gets(t
, &driver
) < 0 ||
1040 pa_tagstruct_getu32(t
, &flags
) < 0) {
1042 pa_log("Parse failure");
1046 if (u
->version
>= 13) {
1047 pa_usec_t configured_latency
;
1049 if (pa_tagstruct_get_proplist(t
, pl
) < 0 ||
1050 pa_tagstruct_get_usec(t
, &configured_latency
) < 0) {
1052 pa_log("Parse failure");
1057 if (u
->version
>= 15) {
1058 pa_volume_t base_volume
;
1059 uint32_t state
, n_volume_steps
, card
;
1061 if (pa_tagstruct_get_volume(t
, &base_volume
) < 0 ||
1062 pa_tagstruct_getu32(t
, &state
) < 0 ||
1063 pa_tagstruct_getu32(t
, &n_volume_steps
) < 0 ||
1064 pa_tagstruct_getu32(t
, &card
) < 0) {
1066 pa_log("Parse failure");
1071 if (u
->version
>= 16) {
1075 if (pa_tagstruct_getu32(t
, &n_ports
)) {
1076 pa_log("Parse failure");
1080 for (uint32_t j
= 0; j
< n_ports
; j
++) {
1083 if (pa_tagstruct_gets(t
, &s
) < 0 || /* name */
1084 pa_tagstruct_gets(t
, &s
) < 0 || /* description */
1085 pa_tagstruct_getu32(t
, &priority
) < 0) {
1087 pa_log("Parse failure");
1092 if (pa_tagstruct_gets(t
, &s
) < 0) { /* active port */
1093 pa_log("Parse failure");
1098 if (!pa_tagstruct_eof(t
)) {
1099 pa_log("Packet too long");
1103 pa_proplist_free(pl
);
1105 if (!u
->sink_name
|| strcmp(name
, u
->sink_name
))
1108 pa_xfree(u
->device_description
);
1109 u
->device_description
= pa_xstrdup(description
);
1111 update_description(u
);
1116 pa_module_unload_request(u
->module
, TRUE
);
1117 pa_proplist_free(pl
);
1120 /* Called from main context */
1121 static void sink_input_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1122 struct userdata
*u
= userdata
;
1123 uint32_t idx
, owner_module
, client
, sink
;
1124 pa_usec_t buffer_usec
, sink_usec
;
1125 const char *name
, *driver
, *resample_method
;
1126 pa_bool_t mute
= FALSE
;
1127 pa_sample_spec sample_spec
;
1128 pa_channel_map channel_map
;
1135 pl
= pa_proplist_new();
1137 if (command
!= PA_COMMAND_REPLY
) {
1138 if (command
== PA_COMMAND_ERROR
)
1139 pa_log("Failed to get info.");
1141 pa_log("Protocol error.");
1145 if (pa_tagstruct_getu32(t
, &idx
) < 0 ||
1146 pa_tagstruct_gets(t
, &name
) < 0 ||
1147 pa_tagstruct_getu32(t
, &owner_module
) < 0 ||
1148 pa_tagstruct_getu32(t
, &client
) < 0 ||
1149 pa_tagstruct_getu32(t
, &sink
) < 0 ||
1150 pa_tagstruct_get_sample_spec(t
, &sample_spec
) < 0 ||
1151 pa_tagstruct_get_channel_map(t
, &channel_map
) < 0 ||
1152 pa_tagstruct_get_cvolume(t
, &volume
) < 0 ||
1153 pa_tagstruct_get_usec(t
, &buffer_usec
) < 0 ||
1154 pa_tagstruct_get_usec(t
, &sink_usec
) < 0 ||
1155 pa_tagstruct_gets(t
, &resample_method
) < 0 ||
1156 pa_tagstruct_gets(t
, &driver
) < 0) {
1158 pa_log("Parse failure");
1162 if (u
->version
>= 11) {
1163 if (pa_tagstruct_get_boolean(t
, &mute
) < 0) {
1165 pa_log("Parse failure");
1170 if (u
->version
>= 13) {
1171 if (pa_tagstruct_get_proplist(t
, pl
) < 0) {
1173 pa_log("Parse failure");
1178 if (!pa_tagstruct_eof(t
)) {
1179 pa_log("Packet too long");
1183 pa_proplist_free(pl
);
1185 if (idx
!= u
->device_index
)
1190 if ((u
->version
< 11 || !!mute
== !!u
->sink
->muted
) &&
1191 pa_cvolume_equal(&volume
, &u
->sink
->real_volume
))
1194 pa_sink_volume_changed(u
->sink
, &volume
);
1196 if (u
->version
>= 11)
1197 pa_sink_mute_changed(u
->sink
, mute
);
1202 pa_module_unload_request(u
->module
, TRUE
);
1203 pa_proplist_free(pl
);
1208 /* Called from main context */
1209 static void source_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1210 struct userdata
*u
= userdata
;
1211 uint32_t idx
, owner_module
, monitor_of_sink
, flags
;
1212 const char *name
, *description
, *monitor_of_sink_name
, *driver
;
1217 pa_usec_t latency
, configured_latency
;
1223 pl
= pa_proplist_new();
1225 if (command
!= PA_COMMAND_REPLY
) {
1226 if (command
== PA_COMMAND_ERROR
)
1227 pa_log("Failed to get info.");
1229 pa_log("Protocol error.");
1233 if (pa_tagstruct_getu32(t
, &idx
) < 0 ||
1234 pa_tagstruct_gets(t
, &name
) < 0 ||
1235 pa_tagstruct_gets(t
, &description
) < 0 ||
1236 pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
1237 pa_tagstruct_get_channel_map(t
, &cm
) < 0 ||
1238 pa_tagstruct_getu32(t
, &owner_module
) < 0 ||
1239 pa_tagstruct_get_cvolume(t
, &volume
) < 0 ||
1240 pa_tagstruct_get_boolean(t
, &mute
) < 0 ||
1241 pa_tagstruct_getu32(t
, &monitor_of_sink
) < 0 ||
1242 pa_tagstruct_gets(t
, &monitor_of_sink_name
) < 0 ||
1243 pa_tagstruct_get_usec(t
, &latency
) < 0 ||
1244 pa_tagstruct_gets(t
, &driver
) < 0 ||
1245 pa_tagstruct_getu32(t
, &flags
) < 0) {
1247 pa_log("Parse failure");
1251 if (u
->version
>= 13) {
1252 if (pa_tagstruct_get_proplist(t
, pl
) < 0 ||
1253 pa_tagstruct_get_usec(t
, &configured_latency
) < 0) {
1255 pa_log("Parse failure");
1260 if (u
->version
>= 15) {
1261 pa_volume_t base_volume
;
1262 uint32_t state
, n_volume_steps
, card
;
1264 if (pa_tagstruct_get_volume(t
, &base_volume
) < 0 ||
1265 pa_tagstruct_getu32(t
, &state
) < 0 ||
1266 pa_tagstruct_getu32(t
, &n_volume_steps
) < 0 ||
1267 pa_tagstruct_getu32(t
, &card
) < 0) {
1269 pa_log("Parse failure");
1274 if (u
->version
>= 16) {
1278 if (pa_tagstruct_getu32(t
, &n_ports
)) {
1279 pa_log("Parse failure");
1283 for (uint32_t j
= 0; j
< n_ports
; j
++) {
1286 if (pa_tagstruct_gets(t
, &s
) < 0 || /* name */
1287 pa_tagstruct_gets(t
, &s
) < 0 || /* description */
1288 pa_tagstruct_getu32(t
, &priority
) < 0) {
1290 pa_log("Parse failure");
1295 if (pa_tagstruct_gets(t
, &s
) < 0) { /* active port */
1296 pa_log("Parse failure");
1301 if (!pa_tagstruct_eof(t
)) {
1302 pa_log("Packet too long");
1306 pa_proplist_free(pl
);
1308 if (!u
->source_name
|| strcmp(name
, u
->source_name
))
1311 pa_xfree(u
->device_description
);
1312 u
->device_description
= pa_xstrdup(description
);
1314 update_description(u
);
1319 pa_module_unload_request(u
->module
, TRUE
);
1320 pa_proplist_free(pl
);
1325 /* Called from main context */
1326 static void request_info(struct userdata
*u
) {
1331 t
= pa_tagstruct_new(NULL
, 0);
1332 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SERVER_INFO
);
1333 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1334 pa_pstream_send_tagstruct(u
->pstream
, t
);
1335 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, server_info_cb
, u
, NULL
);
1338 t
= pa_tagstruct_new(NULL
, 0);
1339 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SINK_INPUT_INFO
);
1340 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1341 pa_tagstruct_putu32(t
, u
->device_index
);
1342 pa_pstream_send_tagstruct(u
->pstream
, t
);
1343 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, sink_input_info_cb
, u
, NULL
);
1346 t
= pa_tagstruct_new(NULL
, 0);
1347 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SINK_INFO
);
1348 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1349 pa_tagstruct_putu32(t
, PA_INVALID_INDEX
);
1350 pa_tagstruct_puts(t
, u
->sink_name
);
1351 pa_pstream_send_tagstruct(u
->pstream
, t
);
1352 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, sink_info_cb
, u
, NULL
);
1355 if (u
->source_name
) {
1356 t
= pa_tagstruct_new(NULL
, 0);
1357 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SOURCE_INFO
);
1358 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1359 pa_tagstruct_putu32(t
, PA_INVALID_INDEX
);
1360 pa_tagstruct_puts(t
, u
->source_name
);
1361 pa_pstream_send_tagstruct(u
->pstream
, t
);
1362 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, source_info_cb
, u
, NULL
);
1367 /* Called from main context */
1368 static void command_subscribe_event(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1369 struct userdata
*u
= userdata
;
1370 pa_subscription_event_type_t e
;
1376 pa_assert(command
== PA_COMMAND_SUBSCRIBE_EVENT
);
1378 if (pa_tagstruct_getu32(t
, &e
) < 0 ||
1379 pa_tagstruct_getu32(t
, &idx
) < 0) {
1380 pa_log("Invalid protocol reply");
1381 pa_module_unload_request(u
->module
, TRUE
);
1385 if (e
!= (PA_SUBSCRIPTION_EVENT_SERVER
|PA_SUBSCRIPTION_EVENT_CHANGE
) &&
1387 e
!= (PA_SUBSCRIPTION_EVENT_SINK_INPUT
|PA_SUBSCRIPTION_EVENT_CHANGE
) &&
1388 e
!= (PA_SUBSCRIPTION_EVENT_SINK
|PA_SUBSCRIPTION_EVENT_CHANGE
)
1390 e
!= (PA_SUBSCRIPTION_EVENT_SOURCE
|PA_SUBSCRIPTION_EVENT_CHANGE
)
1398 /* Called from main context */
1399 static void start_subscribe(struct userdata
*u
) {
1403 t
= pa_tagstruct_new(NULL
, 0);
1404 pa_tagstruct_putu32(t
, PA_COMMAND_SUBSCRIBE
);
1405 pa_tagstruct_putu32(t
, u
->ctag
++);
1406 pa_tagstruct_putu32(t
, PA_SUBSCRIPTION_MASK_SERVER
|
1408 PA_SUBSCRIPTION_MASK_SINK_INPUT
|PA_SUBSCRIPTION_MASK_SINK
1410 PA_SUBSCRIPTION_MASK_SOURCE
1414 pa_pstream_send_tagstruct(u
->pstream
, t
);
1417 /* Called from main context */
1418 static void create_stream_callback(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1419 struct userdata
*u
= userdata
;
1426 pa_assert(u
->pdispatch
== pd
);
1428 if (command
!= PA_COMMAND_REPLY
) {
1429 if (command
== PA_COMMAND_ERROR
)
1430 pa_log("Failed to create stream.");
1432 pa_log("Protocol error.");
1436 if (pa_tagstruct_getu32(t
, &u
->channel
) < 0 ||
1437 pa_tagstruct_getu32(t
, &u
->device_index
) < 0
1439 || pa_tagstruct_getu32(t
, &bytes
) < 0
1444 if (u
->version
>= 9) {
1446 if (pa_tagstruct_getu32(t
, &u
->maxlength
) < 0 ||
1447 pa_tagstruct_getu32(t
, &u
->tlength
) < 0 ||
1448 pa_tagstruct_getu32(t
, &u
->prebuf
) < 0 ||
1449 pa_tagstruct_getu32(t
, &u
->minreq
) < 0)
1452 if (pa_tagstruct_getu32(t
, &u
->maxlength
) < 0 ||
1453 pa_tagstruct_getu32(t
, &u
->fragsize
) < 0)
1458 if (u
->version
>= 12) {
1461 uint32_t device_index
;
1463 pa_bool_t suspended
;
1465 if (pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
1466 pa_tagstruct_get_channel_map(t
, &cm
) < 0 ||
1467 pa_tagstruct_getu32(t
, &device_index
) < 0 ||
1468 pa_tagstruct_gets(t
, &dn
) < 0 ||
1469 pa_tagstruct_get_boolean(t
, &suspended
) < 0)
1473 pa_xfree(u
->sink_name
);
1474 u
->sink_name
= pa_xstrdup(dn
);
1476 pa_xfree(u
->source_name
);
1477 u
->source_name
= pa_xstrdup(dn
);
1481 if (u
->version
>= 13) {
1484 if (pa_tagstruct_get_usec(t
, &usec
) < 0)
1487 /* #ifdef TUNNEL_SINK */
1488 /* pa_sink_set_latency_range(u->sink, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1490 /* pa_source_set_latency_range(u->source, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1494 if (!pa_tagstruct_eof(t
))
1500 pa_assert(!u
->time_event
);
1501 u
->time_event
= pa_core_rttime_new(u
->core
, pa_rtclock_now() + LATENCY_INTERVAL
, timeout_callback
, u
);
1505 pa_log_debug("Stream created.");
1508 pa_asyncmsgq_post(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REQUEST
, NULL
, bytes
, NULL
, NULL
);
1514 pa_log("Invalid reply. (Create stream)");
1517 pa_module_unload_request(u
->module
, TRUE
);
1521 /* Called from main context */
1522 static void setup_complete_callback(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1523 struct userdata
*u
= userdata
;
1524 pa_tagstruct
*reply
;
1525 char name
[256], un
[128], hn
[128];
1532 pa_assert(u
->pdispatch
== pd
);
1534 if (command
!= PA_COMMAND_REPLY
||
1535 pa_tagstruct_getu32(t
, &u
->version
) < 0 ||
1536 !pa_tagstruct_eof(t
)) {
1538 if (command
== PA_COMMAND_ERROR
)
1539 pa_log("Failed to authenticate");
1541 pa_log("Protocol error.");
1546 /* Minimum supported protocol version */
1547 if (u
->version
< 8) {
1548 pa_log("Incompatible protocol version");
1552 /* Starting with protocol version 13 the MSB of the version tag
1553 reflects if shm is enabled for this connection or not. We don't
1554 support SHM here at all, so we just ignore this. */
1556 if (u
->version
>= 13)
1557 u
->version
&= 0x7FFFFFFFU
;
1559 pa_log_debug("Protocol version: remote %u, local %u", u
->version
, PA_PROTOCOL_VERSION
);
1562 pa_proplist_setf(u
->sink
->proplist
, "tunnel.remote_version", "%u", u
->version
);
1563 pa_sink_update_proplist(u
->sink
, 0, NULL
);
1565 pa_snprintf(name
, sizeof(name
), "%s for %s@%s",
1567 pa_get_user_name(un
, sizeof(un
)),
1568 pa_get_host_name(hn
, sizeof(hn
)));
1570 pa_proplist_setf(u
->source
->proplist
, "tunnel.remote_version", "%u", u
->version
);
1571 pa_source_update_proplist(u
->source
, 0, NULL
);
1573 pa_snprintf(name
, sizeof(name
), "%s for %s@%s",
1575 pa_get_user_name(un
, sizeof(un
)),
1576 pa_get_host_name(hn
, sizeof(hn
)));
1579 reply
= pa_tagstruct_new(NULL
, 0);
1580 pa_tagstruct_putu32(reply
, PA_COMMAND_SET_CLIENT_NAME
);
1581 pa_tagstruct_putu32(reply
, u
->ctag
++);
1583 if (u
->version
>= 13) {
1585 pl
= pa_proplist_new();
1586 pa_proplist_sets(pl
, PA_PROP_APPLICATION_ID
, "org.PulseAudio.PulseAudio");
1587 pa_proplist_sets(pl
, PA_PROP_APPLICATION_VERSION
, PACKAGE_VERSION
);
1588 pa_init_proplist(pl
);
1589 pa_tagstruct_put_proplist(reply
, pl
);
1590 pa_proplist_free(pl
);
1592 pa_tagstruct_puts(reply
, "PulseAudio");
1594 pa_pstream_send_tagstruct(u
->pstream
, reply
);
1595 /* We ignore the server's reply here */
1597 reply
= pa_tagstruct_new(NULL
, 0);
1599 if (u
->version
< 13)
1600 /* Only for older PA versions we need to fill in the maxlength */
1601 u
->maxlength
= 4*1024*1024;
1604 u
->tlength
= (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC
* DEFAULT_TLENGTH_MSEC
, &u
->sink
->sample_spec
);
1605 u
->minreq
= (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC
* DEFAULT_MINREQ_MSEC
, &u
->sink
->sample_spec
);
1606 u
->prebuf
= u
->tlength
;
1608 u
->fragsize
= (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC
* DEFAULT_FRAGSIZE_MSEC
, &u
->source
->sample_spec
);
1612 pa_tagstruct_putu32(reply
, PA_COMMAND_CREATE_PLAYBACK_STREAM
);
1613 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
1615 if (u
->version
< 13)
1616 pa_tagstruct_puts(reply
, name
);
1618 pa_tagstruct_put_sample_spec(reply
, &u
->sink
->sample_spec
);
1619 pa_tagstruct_put_channel_map(reply
, &u
->sink
->channel_map
);
1620 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
);
1621 pa_tagstruct_puts(reply
, u
->sink_name
);
1622 pa_tagstruct_putu32(reply
, u
->maxlength
);
1623 pa_tagstruct_put_boolean(reply
, !PA_SINK_IS_OPENED(pa_sink_get_state(u
->sink
)));
1624 pa_tagstruct_putu32(reply
, u
->tlength
);
1625 pa_tagstruct_putu32(reply
, u
->prebuf
);
1626 pa_tagstruct_putu32(reply
, u
->minreq
);
1627 pa_tagstruct_putu32(reply
, 0);
1628 pa_cvolume_reset(&volume
, u
->sink
->sample_spec
.channels
);
1629 pa_tagstruct_put_cvolume(reply
, &volume
);
1631 pa_tagstruct_putu32(reply
, PA_COMMAND_CREATE_RECORD_STREAM
);
1632 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
1634 if (u
->version
< 13)
1635 pa_tagstruct_puts(reply
, name
);
1637 pa_tagstruct_put_sample_spec(reply
, &u
->source
->sample_spec
);
1638 pa_tagstruct_put_channel_map(reply
, &u
->source
->channel_map
);
1639 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
);
1640 pa_tagstruct_puts(reply
, u
->source_name
);
1641 pa_tagstruct_putu32(reply
, u
->maxlength
);
1642 pa_tagstruct_put_boolean(reply
, !PA_SOURCE_IS_OPENED(pa_source_get_state(u
->source
)));
1643 pa_tagstruct_putu32(reply
, u
->fragsize
);
1646 if (u
->version
>= 12) {
1647 pa_tagstruct_put_boolean(reply
, FALSE
); /* no_remap */
1648 pa_tagstruct_put_boolean(reply
, FALSE
); /* no_remix */
1649 pa_tagstruct_put_boolean(reply
, FALSE
); /* fix_format */
1650 pa_tagstruct_put_boolean(reply
, FALSE
); /* fix_rate */
1651 pa_tagstruct_put_boolean(reply
, FALSE
); /* fix_channels */
1652 pa_tagstruct_put_boolean(reply
, TRUE
); /* no_move */
1653 pa_tagstruct_put_boolean(reply
, FALSE
); /* variable_rate */
1656 if (u
->version
>= 13) {
1659 pa_tagstruct_put_boolean(reply
, FALSE
); /* start muted/peak detect*/
1660 pa_tagstruct_put_boolean(reply
, TRUE
); /* adjust_latency */
1662 pl
= pa_proplist_new();
1663 pa_proplist_sets(pl
, PA_PROP_MEDIA_NAME
, name
);
1664 pa_proplist_sets(pl
, PA_PROP_MEDIA_ROLE
, "abstract");
1665 pa_tagstruct_put_proplist(reply
, pl
);
1666 pa_proplist_free(pl
);
1669 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
); /* direct on input */
1673 if (u
->version
>= 14) {
1675 pa_tagstruct_put_boolean(reply
, FALSE
); /* volume_set */
1677 pa_tagstruct_put_boolean(reply
, TRUE
); /* early rquests */
1680 if (u
->version
>= 15) {
1682 pa_tagstruct_put_boolean(reply
, FALSE
); /* muted_set */
1684 pa_tagstruct_put_boolean(reply
, FALSE
); /* don't inhibit auto suspend */
1685 pa_tagstruct_put_boolean(reply
, FALSE
); /* fail on suspend */
1688 pa_pstream_send_tagstruct(u
->pstream
, reply
);
1689 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, create_stream_callback
, u
, NULL
);
1691 pa_log_debug("Connection authenticated, creating stream ...");
1696 pa_module_unload_request(u
->module
, TRUE
);
1699 /* Called from main context */
1700 static void pstream_die_callback(pa_pstream
*p
, void *userdata
) {
1701 struct userdata
*u
= userdata
;
1706 pa_log_warn("Stream died.");
1707 pa_module_unload_request(u
->module
, TRUE
);
1710 /* Called from main context */
1711 static void pstream_packet_callback(pa_pstream
*p
, pa_packet
*packet
, const pa_creds
*creds
, void *userdata
) {
1712 struct userdata
*u
= userdata
;
1718 if (pa_pdispatch_run(u
->pdispatch
, packet
, creds
, u
) < 0) {
1719 pa_log("Invalid packet");
1720 pa_module_unload_request(u
->module
, TRUE
);
1726 /* Called from main context */
1727 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
) {
1728 struct userdata
*u
= userdata
;
1734 if (channel
!= u
->channel
) {
1735 pa_log("Received memory block on bad channel.");
1736 pa_module_unload_request(u
->module
, TRUE
);
1740 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_POST
, PA_UINT_TO_PTR(seek
), offset
, chunk
);
1742 u
->counter_delta
+= (int64_t) chunk
->length
;
1746 /* Called from main context */
1747 static void on_connection(pa_socket_client
*sc
, pa_iochannel
*io
, void *userdata
) {
1748 struct userdata
*u
= userdata
;
1754 pa_assert(u
->client
== sc
);
1756 pa_socket_client_unref(u
->client
);
1760 pa_log("Connection failed: %s", pa_cstrerror(errno
));
1761 pa_module_unload_request(u
->module
, TRUE
);
1765 u
->pstream
= pa_pstream_new(u
->core
->mainloop
, io
, u
->core
->mempool
);
1766 u
->pdispatch
= pa_pdispatch_new(u
->core
->mainloop
, TRUE
, command_table
, PA_COMMAND_MAX
);
1768 pa_pstream_set_die_callback(u
->pstream
, pstream_die_callback
, u
);
1769 pa_pstream_set_recieve_packet_callback(u
->pstream
, pstream_packet_callback
, u
);
1771 pa_pstream_set_recieve_memblock_callback(u
->pstream
, pstream_memblock_callback
, u
);
1774 t
= pa_tagstruct_new(NULL
, 0);
1775 pa_tagstruct_putu32(t
, PA_COMMAND_AUTH
);
1776 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1777 pa_tagstruct_putu32(t
, PA_PROTOCOL_VERSION
);
1779 pa_tagstruct_put_arbitrary(t
, pa_auth_cookie_read(u
->auth_cookie
, PA_NATIVE_COOKIE_LENGTH
), PA_NATIVE_COOKIE_LENGTH
);
1785 if (pa_iochannel_creds_supported(io
))
1786 pa_iochannel_creds_enable(io
);
1788 ucred
.uid
= getuid();
1789 ucred
.gid
= getgid();
1791 pa_pstream_send_tagstruct_with_creds(u
->pstream
, t
, &ucred
);
1794 pa_pstream_send_tagstruct(u
->pstream
, t
);
1797 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, setup_complete_callback
, u
, NULL
);
1799 pa_log_debug("Connection established, authenticating ...");
1804 /* Called from main context */
1805 static void sink_set_volume(pa_sink
*sink
) {
1813 t
= pa_tagstruct_new(NULL
, 0);
1814 pa_tagstruct_putu32(t
, PA_COMMAND_SET_SINK_INPUT_VOLUME
);
1815 pa_tagstruct_putu32(t
, u
->ctag
++);
1816 pa_tagstruct_putu32(t
, u
->device_index
);
1817 pa_tagstruct_put_cvolume(t
, &sink
->real_volume
);
1818 pa_pstream_send_tagstruct(u
->pstream
, t
);
1821 /* Called from main context */
1822 static void sink_set_mute(pa_sink
*sink
) {
1830 if (u
->version
< 11)
1833 t
= pa_tagstruct_new(NULL
, 0);
1834 pa_tagstruct_putu32(t
, PA_COMMAND_SET_SINK_INPUT_MUTE
);
1835 pa_tagstruct_putu32(t
, u
->ctag
++);
1836 pa_tagstruct_putu32(t
, u
->device_index
);
1837 pa_tagstruct_put_boolean(t
, !!sink
->muted
);
1838 pa_pstream_send_tagstruct(u
->pstream
, t
);
1843 int pa__init(pa_module
*m
) {
1844 pa_modargs
*ma
= NULL
;
1845 struct userdata
*u
= NULL
;
1850 pa_sink_new_data data
;
1852 pa_source_new_data data
;
1857 if (!(ma
= pa_modargs_new(m
->argument
, valid_modargs
))) {
1858 pa_log("Failed to parse module arguments");
1862 m
->userdata
= u
= pa_xnew0(struct userdata
, 1);
1866 u
->pdispatch
= NULL
;
1868 u
->server_name
= NULL
;
1870 u
->sink_name
= pa_xstrdup(pa_modargs_get_value(ma
, "sink", NULL
));;
1872 u
->requested_bytes
= 0;
1874 u
->source_name
= pa_xstrdup(pa_modargs_get_value(ma
, "source", NULL
));;
1877 u
->smoother
= pa_smoother_new(
1886 u
->device_index
= u
->channel
= PA_INVALID_INDEX
;
1887 u
->time_event
= NULL
;
1888 u
->ignore_latency_before
= 0;
1889 u
->transport_usec
= u
->thread_transport_usec
= 0;
1890 u
->remote_suspended
= u
->remote_corked
= FALSE
;
1891 u
->counter
= u
->counter_delta
= 0;
1893 u
->rtpoll
= pa_rtpoll_new();
1894 pa_thread_mq_init(&u
->thread_mq
, m
->core
->mainloop
, u
->rtpoll
);
1896 if (!(u
->auth_cookie
= pa_auth_cookie_get(u
->core
, pa_modargs_get_value(ma
, "cookie", PA_NATIVE_COOKIE_FILE
), PA_NATIVE_COOKIE_LENGTH
)))
1899 if (!(u
->server_name
= pa_xstrdup(pa_modargs_get_value(ma
, "server", NULL
)))) {
1900 pa_log("No server specified.");
1904 ss
= m
->core
->default_sample_spec
;
1905 map
= m
->core
->default_channel_map
;
1906 if (pa_modargs_get_sample_spec_and_channel_map(ma
, &ss
, &map
, PA_CHANNEL_MAP_DEFAULT
) < 0) {
1907 pa_log("Invalid sample format specification");
1911 if (!(u
->client
= pa_socket_client_new_string(m
->core
->mainloop
, TRUE
, u
->server_name
, PA_NATIVE_DEFAULT_PORT
))) {
1912 pa_log("Failed to connect to server '%s'", u
->server_name
);
1916 pa_socket_client_set_callback(u
->client
, on_connection
, u
);
1920 if (!(dn
= pa_xstrdup(pa_modargs_get_value(ma
, "sink_name", NULL
))))
1921 dn
= pa_sprintf_malloc("tunnel-sink.%s", u
->server_name
);
1923 pa_sink_new_data_init(&data
);
1924 data
.driver
= __FILE__
;
1926 data
.namereg_fail
= TRUE
;
1927 pa_sink_new_data_set_name(&data
, dn
);
1928 pa_sink_new_data_set_sample_spec(&data
, &ss
);
1929 pa_sink_new_data_set_channel_map(&data
, &map
);
1930 pa_proplist_setf(data
.proplist
, PA_PROP_DEVICE_DESCRIPTION
, "%s%s%s", pa_strempty(u
->sink_name
), u
->sink_name
? " on " : "", u
->server_name
);
1931 pa_proplist_sets(data
.proplist
, "tunnel.remote.server", u
->server_name
);
1933 pa_proplist_sets(data
.proplist
, "tunnel.remote.sink", u
->sink_name
);
1935 if (pa_modargs_get_proplist(ma
, "sink_properties", data
.proplist
, PA_UPDATE_REPLACE
) < 0) {
1936 pa_log("Invalid properties");
1937 pa_sink_new_data_done(&data
);
1941 u
->sink
= pa_sink_new(m
->core
, &data
, PA_SINK_NETWORK
|PA_SINK_LATENCY
|PA_SINK_HW_VOLUME_CTRL
|PA_SINK_HW_MUTE_CTRL
);
1942 pa_sink_new_data_done(&data
);
1945 pa_log("Failed to create sink.");
1949 u
->sink
->parent
.process_msg
= sink_process_msg
;
1950 u
->sink
->userdata
= u
;
1951 u
->sink
->set_state
= sink_set_state
;
1952 u
->sink
->set_volume
= sink_set_volume
;
1953 u
->sink
->set_mute
= sink_set_mute
;
1955 u
->sink
->refresh_volume
= u
->sink
->refresh_muted
= FALSE
;
1957 /* pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
1959 pa_sink_set_asyncmsgq(u
->sink
, u
->thread_mq
.inq
);
1960 pa_sink_set_rtpoll(u
->sink
, u
->rtpoll
);
1964 if (!(dn
= pa_xstrdup(pa_modargs_get_value(ma
, "source_name", NULL
))))
1965 dn
= pa_sprintf_malloc("tunnel-source.%s", u
->server_name
);
1967 pa_source_new_data_init(&data
);
1968 data
.driver
= __FILE__
;
1970 data
.namereg_fail
= TRUE
;
1971 pa_source_new_data_set_name(&data
, dn
);
1972 pa_source_new_data_set_sample_spec(&data
, &ss
);
1973 pa_source_new_data_set_channel_map(&data
, &map
);
1974 pa_proplist_setf(data
.proplist
, PA_PROP_DEVICE_DESCRIPTION
, "%s%s%s", pa_strempty(u
->source_name
), u
->source_name
? " on " : "", u
->server_name
);
1975 pa_proplist_sets(data
.proplist
, "tunnel.remote.server", u
->server_name
);
1977 pa_proplist_sets(data
.proplist
, "tunnel.remote.source", u
->source_name
);
1979 if (pa_modargs_get_proplist(ma
, "source_properties", data
.proplist
, PA_UPDATE_REPLACE
) < 0) {
1980 pa_log("Invalid properties");
1981 pa_source_new_data_done(&data
);
1985 u
->source
= pa_source_new(m
->core
, &data
, PA_SOURCE_NETWORK
|PA_SOURCE_LATENCY
);
1986 pa_source_new_data_done(&data
);
1989 pa_log("Failed to create source.");
1993 u
->source
->parent
.process_msg
= source_process_msg
;
1994 u
->source
->set_state
= source_set_state
;
1995 u
->source
->userdata
= u
;
1997 /* pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
1999 pa_source_set_asyncmsgq(u
->source
, u
->thread_mq
.inq
);
2000 pa_source_set_rtpoll(u
->source
, u
->rtpoll
);
2002 u
->mcalign
= pa_mcalign_new(pa_frame_size(&u
->source
->sample_spec
));
2007 u
->time_event
= NULL
;
2009 u
->maxlength
= (uint32_t) -1;
2011 u
->tlength
= u
->minreq
= u
->prebuf
= (uint32_t) -1;
2013 u
->fragsize
= (uint32_t) -1;
2016 if (!(u
->thread
= pa_thread_new("module-tunnel", thread_func
, u
))) {
2017 pa_log("Failed to create thread.");
2022 pa_sink_put(u
->sink
);
2024 pa_source_put(u
->source
);
2027 pa_modargs_free(ma
);
2035 pa_modargs_free(ma
);
2042 void pa__done(pa_module
*m
) {
2047 if (!(u
= m
->userdata
))
2052 pa_sink_unlink(u
->sink
);
2055 pa_source_unlink(u
->source
);
2059 pa_asyncmsgq_send(u
->thread_mq
.inq
, NULL
, PA_MESSAGE_SHUTDOWN
, NULL
, 0, NULL
);
2060 pa_thread_free(u
->thread
);
2063 pa_thread_mq_done(&u
->thread_mq
);
2067 pa_sink_unref(u
->sink
);
2070 pa_source_unref(u
->source
);
2074 pa_rtpoll_free(u
->rtpoll
);
2077 pa_pstream_unlink(u
->pstream
);
2078 pa_pstream_unref(u
->pstream
);
2082 pa_pdispatch_unref(u
->pdispatch
);
2085 pa_socket_client_unref(u
->client
);
2088 pa_auth_cookie_unref(u
->auth_cookie
);
2091 pa_smoother_free(u
->smoother
);
2094 u
->core
->mainloop
->time_free(u
->time_event
);
2098 pa_mcalign_free(u
->mcalign
);
2102 pa_xfree(u
->sink_name
);
2104 pa_xfree(u
->source_name
);
2106 pa_xfree(u
->server_name
);
2108 pa_xfree(u
->device_description
);
2109 pa_xfree(u
->server_fqdn
);
2110 pa_xfree(u
->user_name
);