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/pdispatch.h>
46 #include <pulsecore/pstream.h>
47 #include <pulsecore/pstream-util.h>
48 #include <pulsecore/socket-client.h>
49 #include <pulsecore/time-smoother.h>
50 #include <pulsecore/thread.h>
51 #include <pulsecore/thread-mq.h>
52 #include <pulsecore/core-rtclock.h>
53 #include <pulsecore/core-error.h>
54 #include <pulsecore/proplist-util.h>
55 #include <pulsecore/auth-cookie.h>
56 #include <pulsecore/mcalign.h>
59 #include "module-tunnel-sink-symdef.h"
61 #include "module-tunnel-source-symdef.h"
65 PA_MODULE_DESCRIPTION("Tunnel module for sinks");
67 "sink_name=<name for the local sink> "
68 "sink_properties=<properties for the local sink> "
70 "sink=<remote sink name> "
72 "format=<sample format> "
73 "channels=<number of channels> "
75 "channel_map=<channel map>");
77 PA_MODULE_DESCRIPTION("Tunnel module for sources");
79 "source_name=<name for the local source> "
80 "source_properties=<properties for the local source> "
82 "source=<remote source name> "
84 "format=<sample format> "
85 "channels=<number of channels> "
87 "channel_map=<channel map>");
90 PA_MODULE_AUTHOR("Lennart Poettering");
91 PA_MODULE_VERSION(PACKAGE_VERSION
);
92 PA_MODULE_LOAD_ONCE(FALSE
);
94 static const char* const valid_modargs
[] = {
113 #define DEFAULT_TIMEOUT 5
115 #define LATENCY_INTERVAL (10*PA_USEC_PER_SEC)
117 #define MIN_NETWORK_LATENCY_USEC (8*PA_USEC_PER_MSEC)
122 SINK_MESSAGE_REQUEST
= PA_SINK_MESSAGE_MAX
,
123 SINK_MESSAGE_REMOTE_SUSPEND
,
124 SINK_MESSAGE_UPDATE_LATENCY
,
128 #define DEFAULT_TLENGTH_MSEC 150
129 #define DEFAULT_MINREQ_MSEC 25
134 SOURCE_MESSAGE_POST
= PA_SOURCE_MESSAGE_MAX
,
135 SOURCE_MESSAGE_REMOTE_SUSPEND
,
136 SOURCE_MESSAGE_UPDATE_LATENCY
139 #define DEFAULT_FRAGSIZE_MSEC 25
144 static void command_request(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
145 static void command_started(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
147 static void command_subscribe_event(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
148 static void command_stream_killed(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
149 static void command_overflow_or_underflow(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
150 static void command_suspended(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
151 static void command_moved(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
152 static void command_stream_or_client_event(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
153 static void command_stream_buffer_attr_changed(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
155 static const pa_pdispatch_cb_t command_table
[PA_COMMAND_MAX
] = {
157 [PA_COMMAND_REQUEST
] = command_request
,
158 [PA_COMMAND_STARTED
] = command_started
,
160 [PA_COMMAND_SUBSCRIBE_EVENT
] = command_subscribe_event
,
161 [PA_COMMAND_OVERFLOW
] = command_overflow_or_underflow
,
162 [PA_COMMAND_UNDERFLOW
] = command_overflow_or_underflow
,
163 [PA_COMMAND_PLAYBACK_STREAM_KILLED
] = command_stream_killed
,
164 [PA_COMMAND_RECORD_STREAM_KILLED
] = command_stream_killed
,
165 [PA_COMMAND_PLAYBACK_STREAM_SUSPENDED
] = command_suspended
,
166 [PA_COMMAND_RECORD_STREAM_SUSPENDED
] = command_suspended
,
167 [PA_COMMAND_PLAYBACK_STREAM_MOVED
] = command_moved
,
168 [PA_COMMAND_RECORD_STREAM_MOVED
] = command_moved
,
169 [PA_COMMAND_PLAYBACK_STREAM_EVENT
] = command_stream_or_client_event
,
170 [PA_COMMAND_RECORD_STREAM_EVENT
] = command_stream_or_client_event
,
171 [PA_COMMAND_CLIENT_EVENT
] = command_stream_or_client_event
,
172 [PA_COMMAND_PLAYBACK_BUFFER_ATTR_CHANGED
] = command_stream_buffer_attr_changed
,
173 [PA_COMMAND_RECORD_BUFFER_ATTR_CHANGED
] = command_stream_buffer_attr_changed
180 pa_thread_mq thread_mq
;
184 pa_socket_client
*client
;
186 pa_pdispatch
*pdispatch
;
192 size_t requested_bytes
;
199 pa_auth_cookie
*auth_cookie
;
203 uint32_t device_index
;
206 int64_t counter
, counter_delta
;
208 pa_bool_t remote_corked
:1;
209 pa_bool_t remote_suspended
:1;
211 pa_usec_t transport_usec
; /* maintained in the main thread */
212 pa_usec_t thread_transport_usec
; /* maintained in the IO thread */
214 uint32_t ignore_latency_before
;
216 pa_time_event
*time_event
;
218 pa_smoother
*smoother
;
220 char *device_description
;
234 static void request_latency(struct userdata
*u
);
236 /* Called from main context */
237 static void command_stream_or_client_event(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
238 pa_log_debug("Got stream or client event.");
241 /* Called from main context */
242 static void command_stream_killed(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
243 struct userdata
*u
= userdata
;
248 pa_assert(u
->pdispatch
== pd
);
250 pa_log_warn("Stream killed");
251 pa_module_unload_request(u
->module
, TRUE
);
254 /* Called from main context */
255 static void command_overflow_or_underflow(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
256 struct userdata
*u
= userdata
;
261 pa_assert(u
->pdispatch
== pd
);
263 pa_log_info("Server signalled buffer overrun/underrun.");
267 /* Called from main context */
268 static void command_suspended(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
269 struct userdata
*u
= userdata
;
276 pa_assert(u
->pdispatch
== pd
);
278 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
279 pa_tagstruct_get_boolean(t
, &suspended
) < 0 ||
280 !pa_tagstruct_eof(t
)) {
282 pa_log("Invalid packet.");
283 pa_module_unload_request(u
->module
, TRUE
);
287 pa_log_debug("Server reports device suspend.");
290 pa_asyncmsgq_send(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
292 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
298 /* Called from main context */
299 static void command_moved(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
300 struct userdata
*u
= userdata
;
301 uint32_t channel
, di
;
308 pa_assert(u
->pdispatch
== pd
);
310 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
311 pa_tagstruct_getu32(t
, &di
) < 0 ||
312 pa_tagstruct_gets(t
, &dn
) < 0 ||
313 pa_tagstruct_get_boolean(t
, &suspended
) < 0) {
315 pa_log_error("Invalid packet.");
316 pa_module_unload_request(u
->module
, TRUE
);
320 pa_log_debug("Server reports a stream move.");
323 pa_asyncmsgq_send(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
325 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_REMOTE_SUSPEND
, PA_UINT32_TO_PTR(!!suspended
), 0, NULL
);
331 static void command_stream_buffer_attr_changed(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
332 struct userdata
*u
= userdata
;
333 uint32_t channel
, maxlength
, tlength
= 0, fragsize
, prebuf
, minreq
;
339 pa_assert(u
->pdispatch
== pd
);
341 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
342 pa_tagstruct_getu32(t
, &maxlength
) < 0) {
344 pa_log_error("Invalid packet.");
345 pa_module_unload_request(u
->module
, TRUE
);
349 if (command
== PA_COMMAND_RECORD_BUFFER_ATTR_CHANGED
) {
350 if (pa_tagstruct_getu32(t
, &fragsize
) < 0 ||
351 pa_tagstruct_get_usec(t
, &usec
) < 0) {
353 pa_log_error("Invalid packet.");
354 pa_module_unload_request(u
->module
, TRUE
);
358 if (pa_tagstruct_getu32(t
, &tlength
) < 0 ||
359 pa_tagstruct_getu32(t
, &prebuf
) < 0 ||
360 pa_tagstruct_getu32(t
, &minreq
) < 0 ||
361 pa_tagstruct_get_usec(t
, &usec
) < 0) {
363 pa_log_error("Invalid packet.");
364 pa_module_unload_request(u
->module
, TRUE
);
370 pa_log_debug("Server reports buffer attrs changed. tlength now at %lu, before %lu.", (unsigned long) tlength
, (unsigned long) u
->tlength
);
378 /* Called from main context */
379 static void command_started(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
380 struct userdata
*u
= userdata
;
385 pa_assert(u
->pdispatch
== pd
);
387 pa_log_debug("Server reports playback started.");
393 /* Called from IO thread context */
394 static void check_smoother_status(struct userdata
*u
, pa_bool_t past
) {
399 x
= pa_rtclock_now();
401 /* Correct by the time the requested issued needs to travel to the
402 * other side. This is a valid thread-safe access, because the
403 * main thread is waiting for us */
406 x
-= u
->thread_transport_usec
;
408 x
+= u
->thread_transport_usec
;
410 if (u
->remote_suspended
|| u
->remote_corked
)
411 pa_smoother_pause(u
->smoother
, x
);
413 pa_smoother_resume(u
->smoother
, x
, TRUE
);
416 /* Called from IO thread context */
417 static void stream_cork_within_thread(struct userdata
*u
, pa_bool_t cork
) {
420 if (u
->remote_corked
== cork
)
423 u
->remote_corked
= cork
;
424 check_smoother_status(u
, FALSE
);
427 /* Called from main context */
428 static void stream_cork(struct userdata
*u
, pa_bool_t cork
) {
435 t
= pa_tagstruct_new(NULL
, 0);
437 pa_tagstruct_putu32(t
, PA_COMMAND_CORK_PLAYBACK_STREAM
);
439 pa_tagstruct_putu32(t
, PA_COMMAND_CORK_RECORD_STREAM
);
441 pa_tagstruct_putu32(t
, u
->ctag
++);
442 pa_tagstruct_putu32(t
, u
->channel
);
443 pa_tagstruct_put_boolean(t
, !!cork
);
444 pa_pstream_send_tagstruct(u
->pstream
, t
);
449 /* Called from IO thread context */
450 static void stream_suspend_within_thread(struct userdata
*u
, pa_bool_t suspend
) {
453 if (u
->remote_suspended
== suspend
)
456 u
->remote_suspended
= suspend
;
457 check_smoother_status(u
, TRUE
);
462 /* Called from IO thread context */
463 static void send_data(struct userdata
*u
) {
466 while (u
->requested_bytes
> 0) {
467 pa_memchunk memchunk
;
469 pa_sink_render(u
->sink
, u
->requested_bytes
, &memchunk
);
470 pa_asyncmsgq_post(u
->thread_mq
.outq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_POST
, NULL
, 0, &memchunk
, NULL
);
471 pa_memblock_unref(memchunk
.memblock
);
473 u
->requested_bytes
-= memchunk
.length
;
475 u
->counter
+= (int64_t) memchunk
.length
;
479 /* This function is called from IO context -- except when it is not. */
480 static int sink_process_msg(pa_msgobject
*o
, int code
, void *data
, int64_t offset
, pa_memchunk
*chunk
) {
481 struct userdata
*u
= PA_SINK(o
)->userdata
;
485 case PA_SINK_MESSAGE_SET_STATE
: {
488 /* First, change the state, because otherwise pa_sink_render() would fail */
489 if ((r
= pa_sink_process_msg(o
, code
, data
, offset
, chunk
)) >= 0) {
491 stream_cork_within_thread(u
, u
->sink
->state
== PA_SINK_SUSPENDED
);
493 if (PA_SINK_IS_OPENED(u
->sink
->state
))
500 case PA_SINK_MESSAGE_GET_LATENCY
: {
501 pa_usec_t yl
, yr
, *usec
= data
;
503 yl
= pa_bytes_to_usec((uint64_t) u
->counter
, &u
->sink
->sample_spec
);
504 yr
= pa_smoother_get(u
->smoother
, pa_rtclock_now());
506 *usec
= yl
> yr
? yl
- yr
: 0;
510 case SINK_MESSAGE_REQUEST
:
512 pa_assert(offset
> 0);
513 u
->requested_bytes
+= (size_t) offset
;
515 if (PA_SINK_IS_OPENED(u
->sink
->thread_info
.state
))
521 case SINK_MESSAGE_REMOTE_SUSPEND
:
523 stream_suspend_within_thread(u
, !!PA_PTR_TO_UINT(data
));
527 case SINK_MESSAGE_UPDATE_LATENCY
: {
530 y
= pa_bytes_to_usec((uint64_t) u
->counter
, &u
->sink
->sample_spec
);
532 if (y
> (pa_usec_t
) offset
)
533 y
-= (pa_usec_t
) offset
;
537 pa_smoother_put(u
->smoother
, pa_rtclock_now(), y
);
539 /* We can access this freely here, since the main thread is waiting for us */
540 u
->thread_transport_usec
= u
->transport_usec
;
545 case SINK_MESSAGE_POST
:
547 /* OK, This might be a bit confusing. This message is
548 * delivered to us from the main context -- NOT from the
549 * IO thread context where the rest of the messages are
550 * dispatched. Yeah, ugly, but I am a lazy bastard. */
552 pa_pstream_send_memblock(u
->pstream
, u
->channel
, 0, PA_SEEK_RELATIVE
, chunk
);
554 u
->counter_delta
+= (int64_t) chunk
->length
;
559 return pa_sink_process_msg(o
, code
, data
, offset
, chunk
);
562 /* Called from main context */
563 static int sink_set_state(pa_sink
*s
, pa_sink_state_t state
) {
565 pa_sink_assert_ref(s
);
568 switch ((pa_sink_state_t
) state
) {
570 case PA_SINK_SUSPENDED
:
571 pa_assert(PA_SINK_IS_OPENED(s
->state
));
572 stream_cork(u
, TRUE
);
576 case PA_SINK_RUNNING
:
577 if (s
->state
== PA_SINK_SUSPENDED
)
578 stream_cork(u
, FALSE
);
581 case PA_SINK_UNLINKED
:
583 case PA_SINK_INVALID_STATE
:
592 /* This function is called from IO context -- except when it is not. */
593 static int source_process_msg(pa_msgobject
*o
, int code
, void *data
, int64_t offset
, pa_memchunk
*chunk
) {
594 struct userdata
*u
= PA_SOURCE(o
)->userdata
;
598 case PA_SOURCE_MESSAGE_SET_STATE
: {
601 if ((r
= pa_source_process_msg(o
, code
, data
, offset
, chunk
)) >= 0)
602 stream_cork_within_thread(u
, u
->source
->state
== PA_SOURCE_SUSPENDED
);
607 case PA_SOURCE_MESSAGE_GET_LATENCY
: {
608 pa_usec_t yr
, yl
, *usec
= data
;
610 yl
= pa_bytes_to_usec((uint64_t) u
->counter
, &PA_SOURCE(o
)->sample_spec
);
611 yr
= pa_smoother_get(u
->smoother
, pa_rtclock_now());
613 *usec
= yr
> yl
? yr
- yl
: 0;
617 case SOURCE_MESSAGE_POST
: {
620 pa_mcalign_push(u
->mcalign
, chunk
);
622 while (pa_mcalign_pop(u
->mcalign
, &c
) >= 0) {
624 if (PA_SOURCE_IS_OPENED(u
->source
->thread_info
.state
))
625 pa_source_post(u
->source
, &c
);
627 pa_memblock_unref(c
.memblock
);
629 u
->counter
+= (int64_t) c
.length
;
635 case SOURCE_MESSAGE_REMOTE_SUSPEND
:
637 stream_suspend_within_thread(u
, !!PA_PTR_TO_UINT(data
));
640 case SOURCE_MESSAGE_UPDATE_LATENCY
: {
643 y
= pa_bytes_to_usec((uint64_t) u
->counter
, &u
->source
->sample_spec
);
644 y
+= (pa_usec_t
) offset
;
646 pa_smoother_put(u
->smoother
, pa_rtclock_now(), y
);
648 /* We can access this freely here, since the main thread is waiting for us */
649 u
->thread_transport_usec
= u
->transport_usec
;
655 return pa_source_process_msg(o
, code
, data
, offset
, chunk
);
658 /* Called from main context */
659 static int source_set_state(pa_source
*s
, pa_source_state_t state
) {
661 pa_source_assert_ref(s
);
664 switch ((pa_source_state_t
) state
) {
666 case PA_SOURCE_SUSPENDED
:
667 pa_assert(PA_SOURCE_IS_OPENED(s
->state
));
668 stream_cork(u
, TRUE
);
672 case PA_SOURCE_RUNNING
:
673 if (s
->state
== PA_SOURCE_SUSPENDED
)
674 stream_cork(u
, FALSE
);
677 case PA_SOURCE_UNLINKED
:
679 case PA_SINK_INVALID_STATE
:
688 static void thread_func(void *userdata
) {
689 struct userdata
*u
= userdata
;
693 pa_log_debug("Thread starting up");
695 pa_thread_mq_install(&u
->thread_mq
);
701 if (PA_UNLIKELY(u
->sink
->thread_info
.rewind_requested
))
702 pa_sink_process_rewind(u
->sink
, 0);
705 if ((ret
= pa_rtpoll_run(u
->rtpoll
, TRUE
)) < 0)
713 /* If this was no regular exit from the loop we have to continue
714 * processing messages until we received PA_MESSAGE_SHUTDOWN */
715 pa_asyncmsgq_post(u
->thread_mq
.outq
, PA_MSGOBJECT(u
->core
), PA_CORE_MESSAGE_UNLOAD_MODULE
, u
->module
, 0, NULL
, NULL
);
716 pa_asyncmsgq_wait_for(u
->thread_mq
.inq
, PA_MESSAGE_SHUTDOWN
);
719 pa_log_debug("Thread shutting down");
723 /* Called from main context */
724 static void command_request(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
725 struct userdata
*u
= userdata
;
726 uint32_t bytes
, channel
;
729 pa_assert(command
== PA_COMMAND_REQUEST
);
732 pa_assert(u
->pdispatch
== pd
);
734 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
735 pa_tagstruct_getu32(t
, &bytes
) < 0) {
736 pa_log("Invalid protocol reply");
740 if (channel
!= u
->channel
) {
741 pa_log("Received data for invalid channel");
745 pa_asyncmsgq_post(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REQUEST
, NULL
, bytes
, NULL
, NULL
);
749 pa_module_unload_request(u
->module
, TRUE
);
754 /* Called from main context */
755 static void stream_get_latency_callback(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
756 struct userdata
*u
= userdata
;
757 pa_usec_t sink_usec
, source_usec
;
759 int64_t write_index
, read_index
;
760 struct timeval local
, remote
, now
;
767 if (command
!= PA_COMMAND_REPLY
) {
768 if (command
== PA_COMMAND_ERROR
)
769 pa_log("Failed to get latency.");
771 pa_log("Protocol error.");
775 if (pa_tagstruct_get_usec(t
, &sink_usec
) < 0 ||
776 pa_tagstruct_get_usec(t
, &source_usec
) < 0 ||
777 pa_tagstruct_get_boolean(t
, &playing
) < 0 ||
778 pa_tagstruct_get_timeval(t
, &local
) < 0 ||
779 pa_tagstruct_get_timeval(t
, &remote
) < 0 ||
780 pa_tagstruct_gets64(t
, &write_index
) < 0 ||
781 pa_tagstruct_gets64(t
, &read_index
) < 0) {
782 pa_log("Invalid reply.");
787 if (u
->version
>= 13) {
788 uint64_t underrun_for
= 0, playing_for
= 0;
790 if (pa_tagstruct_getu64(t
, &underrun_for
) < 0 ||
791 pa_tagstruct_getu64(t
, &playing_for
) < 0) {
792 pa_log("Invalid reply.");
798 if (!pa_tagstruct_eof(t
)) {
799 pa_log("Invalid reply.");
803 if (tag
< u
->ignore_latency_before
) {
807 pa_gettimeofday(&now
);
809 /* Calculate transport usec */
810 if (pa_timeval_cmp(&local
, &remote
) < 0 && pa_timeval_cmp(&remote
, &now
)) {
811 /* local and remote seem to have synchronized clocks */
813 u
->transport_usec
= pa_timeval_diff(&remote
, &local
);
815 u
->transport_usec
= pa_timeval_diff(&now
, &remote
);
818 u
->transport_usec
= pa_timeval_diff(&now
, &local
)/2;
820 /* First, take the device's delay */
822 delay
= (int64_t) sink_usec
;
823 ss
= &u
->sink
->sample_spec
;
825 delay
= (int64_t) source_usec
;
826 ss
= &u
->source
->sample_spec
;
829 /* Add the length of our server-side buffer */
830 if (write_index
>= read_index
)
831 delay
+= (int64_t) pa_bytes_to_usec((uint64_t) (write_index
-read_index
), ss
);
833 delay
-= (int64_t) pa_bytes_to_usec((uint64_t) (read_index
-write_index
), ss
);
835 /* Our measurements are already out of date, hence correct by the *
836 * transport latency */
838 delay
-= (int64_t) u
->transport_usec
;
840 delay
+= (int64_t) u
->transport_usec
;
843 /* Now correct by what we have have read/written since we requested the update */
845 delay
+= (int64_t) pa_bytes_to_usec((uint64_t) u
->counter_delta
, ss
);
847 delay
-= (int64_t) pa_bytes_to_usec((uint64_t) u
->counter_delta
, ss
);
851 pa_asyncmsgq_send(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_UPDATE_LATENCY
, 0, delay
, NULL
);
853 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_UPDATE_LATENCY
, 0, delay
, NULL
);
860 pa_module_unload_request(u
->module
, TRUE
);
863 /* Called from main context */
864 static void request_latency(struct userdata
*u
) {
870 t
= pa_tagstruct_new(NULL
, 0);
872 pa_tagstruct_putu32(t
, PA_COMMAND_GET_PLAYBACK_LATENCY
);
874 pa_tagstruct_putu32(t
, PA_COMMAND_GET_RECORD_LATENCY
);
876 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
877 pa_tagstruct_putu32(t
, u
->channel
);
879 pa_tagstruct_put_timeval(t
, pa_gettimeofday(&now
));
881 pa_pstream_send_tagstruct(u
->pstream
, t
);
882 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, stream_get_latency_callback
, u
, NULL
);
884 u
->ignore_latency_before
= tag
;
885 u
->counter_delta
= 0;
888 /* Called from main context */
889 static void timeout_callback(pa_mainloop_api
*m
, pa_time_event
*e
, const struct timeval
*t
, void *userdata
) {
890 struct userdata
*u
= userdata
;
898 pa_core_rttime_restart(u
->core
, e
, pa_rtclock_now() + LATENCY_INTERVAL
);
901 /* Called from main context */
902 static void update_description(struct userdata
*u
) {
904 char un
[128], hn
[128];
909 if (!u
->server_fqdn
|| !u
->user_name
|| !u
->device_description
)
912 d
= pa_sprintf_malloc("%s on %s@%s", u
->device_description
, u
->user_name
, u
->server_fqdn
);
915 pa_sink_set_description(u
->sink
, d
);
916 pa_proplist_sets(u
->sink
->proplist
, "tunnel.remote.user", u
->user_name
);
917 pa_proplist_sets(u
->sink
->proplist
, "tunnel.remote.fqdn", u
->server_fqdn
);
918 pa_proplist_sets(u
->sink
->proplist
, "tunnel.remote.description", u
->device_description
);
920 pa_source_set_description(u
->source
, d
);
921 pa_proplist_sets(u
->source
->proplist
, "tunnel.remote.user", u
->user_name
);
922 pa_proplist_sets(u
->source
->proplist
, "tunnel.remote.fqdn", u
->server_fqdn
);
923 pa_proplist_sets(u
->source
->proplist
, "tunnel.remote.description", u
->device_description
);
928 d
= pa_sprintf_malloc("%s for %s@%s", u
->device_description
,
929 pa_get_user_name(un
, sizeof(un
)),
930 pa_get_host_name(hn
, sizeof(hn
)));
932 t
= pa_tagstruct_new(NULL
, 0);
934 pa_tagstruct_putu32(t
, PA_COMMAND_SET_PLAYBACK_STREAM_NAME
);
936 pa_tagstruct_putu32(t
, PA_COMMAND_SET_RECORD_STREAM_NAME
);
938 pa_tagstruct_putu32(t
, u
->ctag
++);
939 pa_tagstruct_putu32(t
, u
->channel
);
940 pa_tagstruct_puts(t
, d
);
941 pa_pstream_send_tagstruct(u
->pstream
, t
);
946 /* Called from main context */
947 static void server_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
948 struct userdata
*u
= userdata
;
951 const char *server_name
, *server_version
, *user_name
, *host_name
, *default_sink_name
, *default_source_name
;
957 if (command
!= PA_COMMAND_REPLY
) {
958 if (command
== PA_COMMAND_ERROR
)
959 pa_log("Failed to get info.");
961 pa_log("Protocol error.");
965 if (pa_tagstruct_gets(t
, &server_name
) < 0 ||
966 pa_tagstruct_gets(t
, &server_version
) < 0 ||
967 pa_tagstruct_gets(t
, &user_name
) < 0 ||
968 pa_tagstruct_gets(t
, &host_name
) < 0 ||
969 pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
970 pa_tagstruct_gets(t
, &default_sink_name
) < 0 ||
971 pa_tagstruct_gets(t
, &default_source_name
) < 0 ||
972 pa_tagstruct_getu32(t
, &cookie
) < 0 ||
973 (u
->version
>= 15 && pa_tagstruct_get_channel_map(t
, &cm
) < 0)) {
975 pa_log("Parse failure");
979 if (!pa_tagstruct_eof(t
)) {
980 pa_log("Packet too long");
984 pa_xfree(u
->server_fqdn
);
985 u
->server_fqdn
= pa_xstrdup(host_name
);
987 pa_xfree(u
->user_name
);
988 u
->user_name
= pa_xstrdup(user_name
);
990 update_description(u
);
995 pa_module_unload_request(u
->module
, TRUE
);
998 static int read_ports(struct userdata
*u
, pa_tagstruct
*t
)
1000 if (u
->version
>= 16) {
1004 if (pa_tagstruct_getu32(t
, &n_ports
)) {
1005 pa_log("Parse failure");
1006 return -PA_ERR_PROTOCOL
;
1009 for (uint32_t j
= 0; j
< n_ports
; j
++) {
1012 if (pa_tagstruct_gets(t
, &s
) < 0 || /* name */
1013 pa_tagstruct_gets(t
, &s
) < 0 || /* description */
1014 pa_tagstruct_getu32(t
, &priority
) < 0) {
1016 pa_log("Parse failure");
1017 return -PA_ERR_PROTOCOL
;
1019 if (u
->version
>= 24 && pa_tagstruct_getu32(t
, &priority
) < 0) { /* available */
1020 pa_log("Parse failure");
1021 return -PA_ERR_PROTOCOL
;
1025 if (pa_tagstruct_gets(t
, &s
) < 0) { /* active port */
1026 pa_log("Parse failure");
1027 return -PA_ERR_PROTOCOL
;
1034 static int read_formats(struct userdata
*u
, pa_tagstruct
*t
) {
1036 pa_format_info
*format
;
1038 if (pa_tagstruct_getu8(t
, &n_formats
) < 0) { /* no. of formats */
1039 pa_log("Parse failure");
1040 return -PA_ERR_PROTOCOL
;
1043 for (uint8_t j
= 0; j
< n_formats
; j
++) {
1044 format
= pa_format_info_new();
1045 if (pa_tagstruct_get_format_info(t
, format
)) { /* format info */
1046 pa_format_info_free(format
);
1047 pa_log("Parse failure");
1048 return -PA_ERR_PROTOCOL
;
1050 pa_format_info_free(format
);
1057 /* Called from main context */
1058 static void sink_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1059 struct userdata
*u
= userdata
;
1060 uint32_t idx
, owner_module
, monitor_source
, flags
;
1061 const char *name
, *description
, *monitor_source_name
, *driver
;
1071 if (command
!= PA_COMMAND_REPLY
) {
1072 if (command
== PA_COMMAND_ERROR
)
1073 pa_log("Failed to get info.");
1075 pa_log("Protocol error.");
1079 if (pa_tagstruct_getu32(t
, &idx
) < 0 ||
1080 pa_tagstruct_gets(t
, &name
) < 0 ||
1081 pa_tagstruct_gets(t
, &description
) < 0 ||
1082 pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
1083 pa_tagstruct_get_channel_map(t
, &cm
) < 0 ||
1084 pa_tagstruct_getu32(t
, &owner_module
) < 0 ||
1085 pa_tagstruct_get_cvolume(t
, &volume
) < 0 ||
1086 pa_tagstruct_get_boolean(t
, &mute
) < 0 ||
1087 pa_tagstruct_getu32(t
, &monitor_source
) < 0 ||
1088 pa_tagstruct_gets(t
, &monitor_source_name
) < 0 ||
1089 pa_tagstruct_get_usec(t
, &latency
) < 0 ||
1090 pa_tagstruct_gets(t
, &driver
) < 0 ||
1091 pa_tagstruct_getu32(t
, &flags
) < 0) {
1093 pa_log("Parse failure");
1097 if (u
->version
>= 13) {
1098 pa_usec_t configured_latency
;
1100 if (pa_tagstruct_get_proplist(t
, NULL
) < 0 ||
1101 pa_tagstruct_get_usec(t
, &configured_latency
) < 0) {
1103 pa_log("Parse failure");
1108 if (u
->version
>= 15) {
1109 pa_volume_t base_volume
;
1110 uint32_t state
, n_volume_steps
, card
;
1112 if (pa_tagstruct_get_volume(t
, &base_volume
) < 0 ||
1113 pa_tagstruct_getu32(t
, &state
) < 0 ||
1114 pa_tagstruct_getu32(t
, &n_volume_steps
) < 0 ||
1115 pa_tagstruct_getu32(t
, &card
) < 0) {
1117 pa_log("Parse failure");
1122 if (read_ports(u
, t
) < 0)
1125 if (u
->version
>= 21 && read_formats(u
, t
) < 0)
1128 if (!pa_tagstruct_eof(t
)) {
1129 pa_log("Packet too long");
1133 if (!u
->sink_name
|| !pa_streq(name
, u
->sink_name
))
1136 pa_xfree(u
->device_description
);
1137 u
->device_description
= pa_xstrdup(description
);
1139 update_description(u
);
1144 pa_module_unload_request(u
->module
, TRUE
);
1147 /* Called from main context */
1148 static void sink_input_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1149 struct userdata
*u
= userdata
;
1150 uint32_t idx
, owner_module
, client
, sink
;
1151 pa_usec_t buffer_usec
, sink_usec
;
1152 const char *name
, *driver
, *resample_method
;
1153 pa_bool_t mute
= FALSE
;
1154 pa_sample_spec sample_spec
;
1155 pa_channel_map channel_map
;
1162 if (command
!= PA_COMMAND_REPLY
) {
1163 if (command
== PA_COMMAND_ERROR
)
1164 pa_log("Failed to get info.");
1166 pa_log("Protocol error.");
1170 if (pa_tagstruct_getu32(t
, &idx
) < 0 ||
1171 pa_tagstruct_gets(t
, &name
) < 0 ||
1172 pa_tagstruct_getu32(t
, &owner_module
) < 0 ||
1173 pa_tagstruct_getu32(t
, &client
) < 0 ||
1174 pa_tagstruct_getu32(t
, &sink
) < 0 ||
1175 pa_tagstruct_get_sample_spec(t
, &sample_spec
) < 0 ||
1176 pa_tagstruct_get_channel_map(t
, &channel_map
) < 0 ||
1177 pa_tagstruct_get_cvolume(t
, &volume
) < 0 ||
1178 pa_tagstruct_get_usec(t
, &buffer_usec
) < 0 ||
1179 pa_tagstruct_get_usec(t
, &sink_usec
) < 0 ||
1180 pa_tagstruct_gets(t
, &resample_method
) < 0 ||
1181 pa_tagstruct_gets(t
, &driver
) < 0) {
1183 pa_log("Parse failure");
1187 if (u
->version
>= 11) {
1188 if (pa_tagstruct_get_boolean(t
, &mute
) < 0) {
1190 pa_log("Parse failure");
1195 if (u
->version
>= 13) {
1196 if (pa_tagstruct_get_proplist(t
, NULL
) < 0) {
1198 pa_log("Parse failure");
1203 if (u
->version
>= 19) {
1204 if (pa_tagstruct_get_boolean(t
, &b
) < 0) {
1206 pa_log("Parse failure");
1211 if (u
->version
>= 20) {
1212 if (pa_tagstruct_get_boolean(t
, &b
) < 0 ||
1213 pa_tagstruct_get_boolean(t
, &b
) < 0) {
1215 pa_log("Parse failure");
1220 if (u
->version
>= 21) {
1221 pa_format_info
*format
= pa_format_info_new();
1223 if (pa_tagstruct_get_format_info(t
, format
) < 0) {
1224 pa_format_info_free(format
);
1225 pa_log("Parse failure");
1228 pa_format_info_free(format
);
1231 if (!pa_tagstruct_eof(t
)) {
1232 pa_log("Packet too long");
1236 if (idx
!= u
->device_index
)
1241 if ((u
->version
< 11 || !!mute
== !!u
->sink
->muted
) &&
1242 pa_cvolume_equal(&volume
, &u
->sink
->real_volume
))
1245 pa_sink_volume_changed(u
->sink
, &volume
);
1247 if (u
->version
>= 11)
1248 pa_sink_mute_changed(u
->sink
, mute
);
1253 pa_module_unload_request(u
->module
, TRUE
);
1258 /* Called from main context */
1259 static void source_info_cb(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1260 struct userdata
*u
= userdata
;
1261 uint32_t idx
, owner_module
, monitor_of_sink
, flags
;
1262 const char *name
, *description
, *monitor_of_sink_name
, *driver
;
1267 pa_usec_t latency
, configured_latency
;
1272 if (command
!= PA_COMMAND_REPLY
) {
1273 if (command
== PA_COMMAND_ERROR
)
1274 pa_log("Failed to get info.");
1276 pa_log("Protocol error.");
1280 if (pa_tagstruct_getu32(t
, &idx
) < 0 ||
1281 pa_tagstruct_gets(t
, &name
) < 0 ||
1282 pa_tagstruct_gets(t
, &description
) < 0 ||
1283 pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
1284 pa_tagstruct_get_channel_map(t
, &cm
) < 0 ||
1285 pa_tagstruct_getu32(t
, &owner_module
) < 0 ||
1286 pa_tagstruct_get_cvolume(t
, &volume
) < 0 ||
1287 pa_tagstruct_get_boolean(t
, &mute
) < 0 ||
1288 pa_tagstruct_getu32(t
, &monitor_of_sink
) < 0 ||
1289 pa_tagstruct_gets(t
, &monitor_of_sink_name
) < 0 ||
1290 pa_tagstruct_get_usec(t
, &latency
) < 0 ||
1291 pa_tagstruct_gets(t
, &driver
) < 0 ||
1292 pa_tagstruct_getu32(t
, &flags
) < 0) {
1294 pa_log("Parse failure");
1298 if (u
->version
>= 13) {
1299 if (pa_tagstruct_get_proplist(t
, NULL
) < 0 ||
1300 pa_tagstruct_get_usec(t
, &configured_latency
) < 0) {
1302 pa_log("Parse failure");
1307 if (u
->version
>= 15) {
1308 pa_volume_t base_volume
;
1309 uint32_t state
, n_volume_steps
, card
;
1311 if (pa_tagstruct_get_volume(t
, &base_volume
) < 0 ||
1312 pa_tagstruct_getu32(t
, &state
) < 0 ||
1313 pa_tagstruct_getu32(t
, &n_volume_steps
) < 0 ||
1314 pa_tagstruct_getu32(t
, &card
) < 0) {
1316 pa_log("Parse failure");
1321 if (read_ports(u
, t
) < 0)
1324 if (u
->version
>= 22 && read_formats(u
, t
) < 0)
1327 if (!pa_tagstruct_eof(t
)) {
1328 pa_log("Packet too long");
1332 if (!u
->source_name
|| !pa_streq(name
, u
->source_name
))
1335 pa_xfree(u
->device_description
);
1336 u
->device_description
= pa_xstrdup(description
);
1338 update_description(u
);
1343 pa_module_unload_request(u
->module
, TRUE
);
1348 /* Called from main context */
1349 static void request_info(struct userdata
*u
) {
1354 t
= pa_tagstruct_new(NULL
, 0);
1355 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SERVER_INFO
);
1356 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1357 pa_pstream_send_tagstruct(u
->pstream
, t
);
1358 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, server_info_cb
, u
, NULL
);
1361 t
= pa_tagstruct_new(NULL
, 0);
1362 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SINK_INPUT_INFO
);
1363 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1364 pa_tagstruct_putu32(t
, u
->device_index
);
1365 pa_pstream_send_tagstruct(u
->pstream
, t
);
1366 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, sink_input_info_cb
, u
, NULL
);
1369 t
= pa_tagstruct_new(NULL
, 0);
1370 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SINK_INFO
);
1371 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1372 pa_tagstruct_putu32(t
, PA_INVALID_INDEX
);
1373 pa_tagstruct_puts(t
, u
->sink_name
);
1374 pa_pstream_send_tagstruct(u
->pstream
, t
);
1375 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, sink_info_cb
, u
, NULL
);
1378 if (u
->source_name
) {
1379 t
= pa_tagstruct_new(NULL
, 0);
1380 pa_tagstruct_putu32(t
, PA_COMMAND_GET_SOURCE_INFO
);
1381 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1382 pa_tagstruct_putu32(t
, PA_INVALID_INDEX
);
1383 pa_tagstruct_puts(t
, u
->source_name
);
1384 pa_pstream_send_tagstruct(u
->pstream
, t
);
1385 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, source_info_cb
, u
, NULL
);
1390 /* Called from main context */
1391 static void command_subscribe_event(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1392 struct userdata
*u
= userdata
;
1393 pa_subscription_event_type_t e
;
1399 pa_assert(command
== PA_COMMAND_SUBSCRIBE_EVENT
);
1401 if (pa_tagstruct_getu32(t
, &e
) < 0 ||
1402 pa_tagstruct_getu32(t
, &idx
) < 0) {
1403 pa_log("Invalid protocol reply");
1404 pa_module_unload_request(u
->module
, TRUE
);
1408 if (e
!= (PA_SUBSCRIPTION_EVENT_SERVER
|PA_SUBSCRIPTION_EVENT_CHANGE
) &&
1410 e
!= (PA_SUBSCRIPTION_EVENT_SINK_INPUT
|PA_SUBSCRIPTION_EVENT_CHANGE
) &&
1411 e
!= (PA_SUBSCRIPTION_EVENT_SINK
|PA_SUBSCRIPTION_EVENT_CHANGE
)
1413 e
!= (PA_SUBSCRIPTION_EVENT_SOURCE
|PA_SUBSCRIPTION_EVENT_CHANGE
)
1421 /* Called from main context */
1422 static void start_subscribe(struct userdata
*u
) {
1426 t
= pa_tagstruct_new(NULL
, 0);
1427 pa_tagstruct_putu32(t
, PA_COMMAND_SUBSCRIBE
);
1428 pa_tagstruct_putu32(t
, u
->ctag
++);
1429 pa_tagstruct_putu32(t
, PA_SUBSCRIPTION_MASK_SERVER
|
1431 PA_SUBSCRIPTION_MASK_SINK_INPUT
|PA_SUBSCRIPTION_MASK_SINK
1433 PA_SUBSCRIPTION_MASK_SOURCE
1437 pa_pstream_send_tagstruct(u
->pstream
, t
);
1440 /* Called from main context */
1441 static void create_stream_callback(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1442 struct userdata
*u
= userdata
;
1449 pa_assert(u
->pdispatch
== pd
);
1451 if (command
!= PA_COMMAND_REPLY
) {
1452 if (command
== PA_COMMAND_ERROR
)
1453 pa_log("Failed to create stream.");
1455 pa_log("Protocol error.");
1459 if (pa_tagstruct_getu32(t
, &u
->channel
) < 0 ||
1460 pa_tagstruct_getu32(t
, &u
->device_index
) < 0
1462 || pa_tagstruct_getu32(t
, &bytes
) < 0
1467 if (u
->version
>= 9) {
1469 if (pa_tagstruct_getu32(t
, &u
->maxlength
) < 0 ||
1470 pa_tagstruct_getu32(t
, &u
->tlength
) < 0 ||
1471 pa_tagstruct_getu32(t
, &u
->prebuf
) < 0 ||
1472 pa_tagstruct_getu32(t
, &u
->minreq
) < 0)
1475 if (pa_tagstruct_getu32(t
, &u
->maxlength
) < 0 ||
1476 pa_tagstruct_getu32(t
, &u
->fragsize
) < 0)
1481 if (u
->version
>= 12) {
1484 uint32_t device_index
;
1486 pa_bool_t suspended
;
1488 if (pa_tagstruct_get_sample_spec(t
, &ss
) < 0 ||
1489 pa_tagstruct_get_channel_map(t
, &cm
) < 0 ||
1490 pa_tagstruct_getu32(t
, &device_index
) < 0 ||
1491 pa_tagstruct_gets(t
, &dn
) < 0 ||
1492 pa_tagstruct_get_boolean(t
, &suspended
) < 0)
1496 pa_xfree(u
->sink_name
);
1497 u
->sink_name
= pa_xstrdup(dn
);
1499 pa_xfree(u
->source_name
);
1500 u
->source_name
= pa_xstrdup(dn
);
1504 if (u
->version
>= 13) {
1507 if (pa_tagstruct_get_usec(t
, &usec
) < 0)
1510 /* #ifdef TUNNEL_SINK */
1511 /* pa_sink_set_latency_range(u->sink, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1513 /* pa_source_set_latency_range(u->source, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1517 if (u
->version
>= 21) {
1518 pa_format_info
*format
= pa_format_info_new();
1520 if (pa_tagstruct_get_format_info(t
, format
) < 0) {
1521 pa_format_info_free(format
);
1525 pa_format_info_free(format
);
1528 if (!pa_tagstruct_eof(t
))
1534 pa_assert(!u
->time_event
);
1535 u
->time_event
= pa_core_rttime_new(u
->core
, pa_rtclock_now() + LATENCY_INTERVAL
, timeout_callback
, u
);
1539 pa_log_debug("Stream created.");
1542 pa_asyncmsgq_post(u
->sink
->asyncmsgq
, PA_MSGOBJECT(u
->sink
), SINK_MESSAGE_REQUEST
, NULL
, bytes
, NULL
, NULL
);
1548 pa_log("Invalid reply. (Create stream)");
1551 pa_module_unload_request(u
->module
, TRUE
);
1555 /* Called from main context */
1556 static void setup_complete_callback(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
1557 struct userdata
*u
= userdata
;
1558 pa_tagstruct
*reply
;
1559 char name
[256], un
[128], hn
[128];
1564 pa_assert(u
->pdispatch
== pd
);
1566 if (command
!= PA_COMMAND_REPLY
||
1567 pa_tagstruct_getu32(t
, &u
->version
) < 0 ||
1568 !pa_tagstruct_eof(t
)) {
1570 if (command
== PA_COMMAND_ERROR
)
1571 pa_log("Failed to authenticate");
1573 pa_log("Protocol error.");
1578 /* Minimum supported protocol version */
1579 if (u
->version
< 8) {
1580 pa_log("Incompatible protocol version");
1584 /* Starting with protocol version 13 the MSB of the version tag
1585 reflects if shm is enabled for this connection or not. We don't
1586 support SHM here at all, so we just ignore this. */
1588 if (u
->version
>= 13)
1589 u
->version
&= 0x7FFFFFFFU
;
1591 pa_log_debug("Protocol version: remote %u, local %u", u
->version
, PA_PROTOCOL_VERSION
);
1594 pa_proplist_setf(u
->sink
->proplist
, "tunnel.remote_version", "%u", u
->version
);
1595 pa_sink_update_proplist(u
->sink
, 0, NULL
);
1597 pa_snprintf(name
, sizeof(name
), "%s for %s@%s",
1599 pa_get_user_name(un
, sizeof(un
)),
1600 pa_get_host_name(hn
, sizeof(hn
)));
1602 pa_proplist_setf(u
->source
->proplist
, "tunnel.remote_version", "%u", u
->version
);
1603 pa_source_update_proplist(u
->source
, 0, NULL
);
1605 pa_snprintf(name
, sizeof(name
), "%s for %s@%s",
1607 pa_get_user_name(un
, sizeof(un
)),
1608 pa_get_host_name(hn
, sizeof(hn
)));
1611 reply
= pa_tagstruct_new(NULL
, 0);
1612 pa_tagstruct_putu32(reply
, PA_COMMAND_SET_CLIENT_NAME
);
1613 pa_tagstruct_putu32(reply
, u
->ctag
++);
1615 if (u
->version
>= 13) {
1617 pl
= pa_proplist_new();
1618 pa_proplist_sets(pl
, PA_PROP_APPLICATION_ID
, "org.PulseAudio.PulseAudio");
1619 pa_proplist_sets(pl
, PA_PROP_APPLICATION_VERSION
, PACKAGE_VERSION
);
1620 pa_init_proplist(pl
);
1621 pa_tagstruct_put_proplist(reply
, pl
);
1622 pa_proplist_free(pl
);
1624 pa_tagstruct_puts(reply
, "PulseAudio");
1626 pa_pstream_send_tagstruct(u
->pstream
, reply
);
1627 /* We ignore the server's reply here */
1629 reply
= pa_tagstruct_new(NULL
, 0);
1631 if (u
->version
< 13)
1632 /* Only for older PA versions we need to fill in the maxlength */
1633 u
->maxlength
= 4*1024*1024;
1636 u
->tlength
= (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC
* DEFAULT_TLENGTH_MSEC
, &u
->sink
->sample_spec
);
1637 u
->minreq
= (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC
* DEFAULT_MINREQ_MSEC
, &u
->sink
->sample_spec
);
1638 u
->prebuf
= u
->tlength
;
1640 u
->fragsize
= (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC
* DEFAULT_FRAGSIZE_MSEC
, &u
->source
->sample_spec
);
1644 pa_tagstruct_putu32(reply
, PA_COMMAND_CREATE_PLAYBACK_STREAM
);
1645 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
1647 if (u
->version
< 13)
1648 pa_tagstruct_puts(reply
, name
);
1650 pa_tagstruct_put_sample_spec(reply
, &u
->sink
->sample_spec
);
1651 pa_tagstruct_put_channel_map(reply
, &u
->sink
->channel_map
);
1652 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
);
1653 pa_tagstruct_puts(reply
, u
->sink_name
);
1654 pa_tagstruct_putu32(reply
, u
->maxlength
);
1655 pa_tagstruct_put_boolean(reply
, !PA_SINK_IS_OPENED(pa_sink_get_state(u
->sink
)));
1656 pa_tagstruct_putu32(reply
, u
->tlength
);
1657 pa_tagstruct_putu32(reply
, u
->prebuf
);
1658 pa_tagstruct_putu32(reply
, u
->minreq
);
1659 pa_tagstruct_putu32(reply
, 0);
1660 pa_cvolume_reset(&volume
, u
->sink
->sample_spec
.channels
);
1661 pa_tagstruct_put_cvolume(reply
, &volume
);
1663 pa_tagstruct_putu32(reply
, PA_COMMAND_CREATE_RECORD_STREAM
);
1664 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
1666 if (u
->version
< 13)
1667 pa_tagstruct_puts(reply
, name
);
1669 pa_tagstruct_put_sample_spec(reply
, &u
->source
->sample_spec
);
1670 pa_tagstruct_put_channel_map(reply
, &u
->source
->channel_map
);
1671 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
);
1672 pa_tagstruct_puts(reply
, u
->source_name
);
1673 pa_tagstruct_putu32(reply
, u
->maxlength
);
1674 pa_tagstruct_put_boolean(reply
, !PA_SOURCE_IS_OPENED(pa_source_get_state(u
->source
)));
1675 pa_tagstruct_putu32(reply
, u
->fragsize
);
1678 if (u
->version
>= 12) {
1679 pa_tagstruct_put_boolean(reply
, FALSE
); /* no_remap */
1680 pa_tagstruct_put_boolean(reply
, FALSE
); /* no_remix */
1681 pa_tagstruct_put_boolean(reply
, FALSE
); /* fix_format */
1682 pa_tagstruct_put_boolean(reply
, FALSE
); /* fix_rate */
1683 pa_tagstruct_put_boolean(reply
, FALSE
); /* fix_channels */
1684 pa_tagstruct_put_boolean(reply
, TRUE
); /* no_move */
1685 pa_tagstruct_put_boolean(reply
, FALSE
); /* variable_rate */
1688 if (u
->version
>= 13) {
1691 pa_tagstruct_put_boolean(reply
, FALSE
); /* start muted/peak detect*/
1692 pa_tagstruct_put_boolean(reply
, TRUE
); /* adjust_latency */
1694 pl
= pa_proplist_new();
1695 pa_proplist_sets(pl
, PA_PROP_MEDIA_NAME
, name
);
1696 pa_proplist_sets(pl
, PA_PROP_MEDIA_ROLE
, "abstract");
1697 pa_tagstruct_put_proplist(reply
, pl
);
1698 pa_proplist_free(pl
);
1701 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
); /* direct on input */
1705 if (u
->version
>= 14) {
1707 pa_tagstruct_put_boolean(reply
, FALSE
); /* volume_set */
1709 pa_tagstruct_put_boolean(reply
, TRUE
); /* early rquests */
1712 if (u
->version
>= 15) {
1714 pa_tagstruct_put_boolean(reply
, FALSE
); /* muted_set */
1716 pa_tagstruct_put_boolean(reply
, FALSE
); /* don't inhibit auto suspend */
1717 pa_tagstruct_put_boolean(reply
, FALSE
); /* fail on suspend */
1721 if (u
->version
>= 17)
1722 pa_tagstruct_put_boolean(reply
, FALSE
); /* relative volume */
1724 if (u
->version
>= 18)
1725 pa_tagstruct_put_boolean(reply
, FALSE
); /* passthrough stream */
1729 if (u
->version
>= 21) {
1730 /* We're not using the extended API, so n_formats = 0 and that's that */
1731 pa_tagstruct_putu8(reply
, 0);
1734 if (u
->version
>= 22) {
1735 /* We're not using the extended API, so n_formats = 0 and that's that */
1736 pa_tagstruct_putu8(reply
, 0);
1737 pa_cvolume_reset(&volume
, u
->source
->sample_spec
.channels
);
1738 pa_tagstruct_put_cvolume(reply
, &volume
);
1739 pa_tagstruct_put_boolean(reply
, FALSE
); /* muted */
1740 pa_tagstruct_put_boolean(reply
, FALSE
); /* volume_set */
1741 pa_tagstruct_put_boolean(reply
, FALSE
); /* muted_set */
1742 pa_tagstruct_put_boolean(reply
, FALSE
); /* relative volume */
1743 pa_tagstruct_put_boolean(reply
, FALSE
); /* passthrough stream */
1747 pa_pstream_send_tagstruct(u
->pstream
, reply
);
1748 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, create_stream_callback
, u
, NULL
);
1750 pa_log_debug("Connection authenticated, creating stream ...");
1755 pa_module_unload_request(u
->module
, TRUE
);
1758 /* Called from main context */
1759 static void pstream_die_callback(pa_pstream
*p
, void *userdata
) {
1760 struct userdata
*u
= userdata
;
1765 pa_log_warn("Stream died.");
1766 pa_module_unload_request(u
->module
, TRUE
);
1769 /* Called from main context */
1770 static void pstream_packet_callback(pa_pstream
*p
, pa_packet
*packet
, const pa_creds
*creds
, void *userdata
) {
1771 struct userdata
*u
= userdata
;
1777 if (pa_pdispatch_run(u
->pdispatch
, packet
, creds
, u
) < 0) {
1778 pa_log("Invalid packet");
1779 pa_module_unload_request(u
->module
, TRUE
);
1785 /* Called from main context */
1786 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
) {
1787 struct userdata
*u
= userdata
;
1793 if (channel
!= u
->channel
) {
1794 pa_log("Received memory block on bad channel.");
1795 pa_module_unload_request(u
->module
, TRUE
);
1799 pa_asyncmsgq_send(u
->source
->asyncmsgq
, PA_MSGOBJECT(u
->source
), SOURCE_MESSAGE_POST
, PA_UINT_TO_PTR(seek
), offset
, chunk
);
1801 u
->counter_delta
+= (int64_t) chunk
->length
;
1805 /* Called from main context */
1806 static void on_connection(pa_socket_client
*sc
, pa_iochannel
*io
, void *userdata
) {
1807 struct userdata
*u
= userdata
;
1813 pa_assert(u
->client
== sc
);
1815 pa_socket_client_unref(u
->client
);
1819 pa_log("Connection failed: %s", pa_cstrerror(errno
));
1820 pa_module_unload_request(u
->module
, TRUE
);
1824 u
->pstream
= pa_pstream_new(u
->core
->mainloop
, io
, u
->core
->mempool
);
1825 u
->pdispatch
= pa_pdispatch_new(u
->core
->mainloop
, TRUE
, command_table
, PA_COMMAND_MAX
);
1827 pa_pstream_set_die_callback(u
->pstream
, pstream_die_callback
, u
);
1828 pa_pstream_set_receive_packet_callback(u
->pstream
, pstream_packet_callback
, u
);
1830 pa_pstream_set_receive_memblock_callback(u
->pstream
, pstream_memblock_callback
, u
);
1833 t
= pa_tagstruct_new(NULL
, 0);
1834 pa_tagstruct_putu32(t
, PA_COMMAND_AUTH
);
1835 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
1836 pa_tagstruct_putu32(t
, PA_PROTOCOL_VERSION
);
1838 pa_tagstruct_put_arbitrary(t
, pa_auth_cookie_read(u
->auth_cookie
, PA_NATIVE_COOKIE_LENGTH
), PA_NATIVE_COOKIE_LENGTH
);
1844 if (pa_iochannel_creds_supported(io
))
1845 pa_iochannel_creds_enable(io
);
1847 ucred
.uid
= getuid();
1848 ucred
.gid
= getgid();
1850 pa_pstream_send_tagstruct_with_creds(u
->pstream
, t
, &ucred
);
1853 pa_pstream_send_tagstruct(u
->pstream
, t
);
1856 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, setup_complete_callback
, u
, NULL
);
1858 pa_log_debug("Connection established, authenticating ...");
1863 /* Called from main context */
1864 static void sink_set_volume(pa_sink
*sink
) {
1872 t
= pa_tagstruct_new(NULL
, 0);
1873 pa_tagstruct_putu32(t
, PA_COMMAND_SET_SINK_INPUT_VOLUME
);
1874 pa_tagstruct_putu32(t
, u
->ctag
++);
1875 pa_tagstruct_putu32(t
, u
->device_index
);
1876 pa_tagstruct_put_cvolume(t
, &sink
->real_volume
);
1877 pa_pstream_send_tagstruct(u
->pstream
, t
);
1880 /* Called from main context */
1881 static void sink_set_mute(pa_sink
*sink
) {
1889 if (u
->version
< 11)
1892 t
= pa_tagstruct_new(NULL
, 0);
1893 pa_tagstruct_putu32(t
, PA_COMMAND_SET_SINK_INPUT_MUTE
);
1894 pa_tagstruct_putu32(t
, u
->ctag
++);
1895 pa_tagstruct_putu32(t
, u
->device_index
);
1896 pa_tagstruct_put_boolean(t
, !!sink
->muted
);
1897 pa_pstream_send_tagstruct(u
->pstream
, t
);
1902 int pa__init(pa_module
*m
) {
1903 pa_modargs
*ma
= NULL
;
1904 struct userdata
*u
= NULL
;
1909 pa_sink_new_data data
;
1911 pa_source_new_data data
;
1916 if (!(ma
= pa_modargs_new(m
->argument
, valid_modargs
))) {
1917 pa_log("Failed to parse module arguments");
1921 m
->userdata
= u
= pa_xnew0(struct userdata
, 1);
1925 u
->pdispatch
= NULL
;
1927 u
->server_name
= NULL
;
1929 u
->sink_name
= pa_xstrdup(pa_modargs_get_value(ma
, "sink", NULL
));;
1931 u
->requested_bytes
= 0;
1933 u
->source_name
= pa_xstrdup(pa_modargs_get_value(ma
, "source", NULL
));;
1936 u
->smoother
= pa_smoother_new(
1945 u
->device_index
= u
->channel
= PA_INVALID_INDEX
;
1946 u
->time_event
= NULL
;
1947 u
->ignore_latency_before
= 0;
1948 u
->transport_usec
= u
->thread_transport_usec
= 0;
1949 u
->remote_suspended
= u
->remote_corked
= FALSE
;
1950 u
->counter
= u
->counter_delta
= 0;
1952 u
->rtpoll
= pa_rtpoll_new();
1953 pa_thread_mq_init(&u
->thread_mq
, m
->core
->mainloop
, u
->rtpoll
);
1955 if (!(u
->auth_cookie
= pa_auth_cookie_get(u
->core
, pa_modargs_get_value(ma
, "cookie", PA_NATIVE_COOKIE_FILE
), TRUE
, PA_NATIVE_COOKIE_LENGTH
)))
1958 if (!(u
->server_name
= pa_xstrdup(pa_modargs_get_value(ma
, "server", NULL
)))) {
1959 pa_log("No server specified.");
1963 ss
= m
->core
->default_sample_spec
;
1964 map
= m
->core
->default_channel_map
;
1965 if (pa_modargs_get_sample_spec_and_channel_map(ma
, &ss
, &map
, PA_CHANNEL_MAP_DEFAULT
) < 0) {
1966 pa_log("Invalid sample format specification");
1970 if (!(u
->client
= pa_socket_client_new_string(m
->core
->mainloop
, TRUE
, u
->server_name
, PA_NATIVE_DEFAULT_PORT
))) {
1971 pa_log("Failed to connect to server '%s'", u
->server_name
);
1975 pa_socket_client_set_callback(u
->client
, on_connection
, u
);
1979 if (!(dn
= pa_xstrdup(pa_modargs_get_value(ma
, "sink_name", NULL
))))
1980 dn
= pa_sprintf_malloc("tunnel-sink.%s", u
->server_name
);
1982 pa_sink_new_data_init(&data
);
1983 data
.driver
= __FILE__
;
1985 data
.namereg_fail
= TRUE
;
1986 pa_sink_new_data_set_name(&data
, dn
);
1987 pa_sink_new_data_set_sample_spec(&data
, &ss
);
1988 pa_sink_new_data_set_channel_map(&data
, &map
);
1989 pa_proplist_setf(data
.proplist
, PA_PROP_DEVICE_DESCRIPTION
, "%s%s%s", pa_strempty(u
->sink_name
), u
->sink_name
? " on " : "", u
->server_name
);
1990 pa_proplist_sets(data
.proplist
, "tunnel.remote.server", u
->server_name
);
1992 pa_proplist_sets(data
.proplist
, "tunnel.remote.sink", u
->sink_name
);
1994 if (pa_modargs_get_proplist(ma
, "sink_properties", data
.proplist
, PA_UPDATE_REPLACE
) < 0) {
1995 pa_log("Invalid properties");
1996 pa_sink_new_data_done(&data
);
2000 u
->sink
= pa_sink_new(m
->core
, &data
, PA_SINK_NETWORK
|PA_SINK_LATENCY
);
2001 pa_sink_new_data_done(&data
);
2004 pa_log("Failed to create sink.");
2008 u
->sink
->parent
.process_msg
= sink_process_msg
;
2009 u
->sink
->userdata
= u
;
2010 u
->sink
->set_state
= sink_set_state
;
2011 pa_sink_set_set_volume_callback(u
->sink
, sink_set_volume
);
2012 pa_sink_set_set_mute_callback(u
->sink
, sink_set_mute
);
2014 u
->sink
->refresh_volume
= u
->sink
->refresh_muted
= FALSE
;
2016 /* pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
2018 pa_sink_set_asyncmsgq(u
->sink
, u
->thread_mq
.inq
);
2019 pa_sink_set_rtpoll(u
->sink
, u
->rtpoll
);
2023 if (!(dn
= pa_xstrdup(pa_modargs_get_value(ma
, "source_name", NULL
))))
2024 dn
= pa_sprintf_malloc("tunnel-source.%s", u
->server_name
);
2026 pa_source_new_data_init(&data
);
2027 data
.driver
= __FILE__
;
2029 data
.namereg_fail
= TRUE
;
2030 pa_source_new_data_set_name(&data
, dn
);
2031 pa_source_new_data_set_sample_spec(&data
, &ss
);
2032 pa_source_new_data_set_channel_map(&data
, &map
);
2033 pa_proplist_setf(data
.proplist
, PA_PROP_DEVICE_DESCRIPTION
, "%s%s%s", pa_strempty(u
->source_name
), u
->source_name
? " on " : "", u
->server_name
);
2034 pa_proplist_sets(data
.proplist
, "tunnel.remote.server", u
->server_name
);
2036 pa_proplist_sets(data
.proplist
, "tunnel.remote.source", u
->source_name
);
2038 if (pa_modargs_get_proplist(ma
, "source_properties", data
.proplist
, PA_UPDATE_REPLACE
) < 0) {
2039 pa_log("Invalid properties");
2040 pa_source_new_data_done(&data
);
2044 u
->source
= pa_source_new(m
->core
, &data
, PA_SOURCE_NETWORK
|PA_SOURCE_LATENCY
);
2045 pa_source_new_data_done(&data
);
2048 pa_log("Failed to create source.");
2052 u
->source
->parent
.process_msg
= source_process_msg
;
2053 u
->source
->set_state
= source_set_state
;
2054 u
->source
->userdata
= u
;
2056 /* pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
2058 pa_source_set_asyncmsgq(u
->source
, u
->thread_mq
.inq
);
2059 pa_source_set_rtpoll(u
->source
, u
->rtpoll
);
2061 u
->mcalign
= pa_mcalign_new(pa_frame_size(&u
->source
->sample_spec
));
2066 u
->time_event
= NULL
;
2068 u
->maxlength
= (uint32_t) -1;
2070 u
->tlength
= u
->minreq
= u
->prebuf
= (uint32_t) -1;
2072 u
->fragsize
= (uint32_t) -1;
2075 if (!(u
->thread
= pa_thread_new("module-tunnel", thread_func
, u
))) {
2076 pa_log("Failed to create thread.");
2081 pa_sink_put(u
->sink
);
2083 pa_source_put(u
->source
);
2086 pa_modargs_free(ma
);
2094 pa_modargs_free(ma
);
2101 void pa__done(pa_module
*m
) {
2106 if (!(u
= m
->userdata
))
2111 pa_sink_unlink(u
->sink
);
2114 pa_source_unlink(u
->source
);
2118 pa_asyncmsgq_send(u
->thread_mq
.inq
, NULL
, PA_MESSAGE_SHUTDOWN
, NULL
, 0, NULL
);
2119 pa_thread_free(u
->thread
);
2122 pa_thread_mq_done(&u
->thread_mq
);
2126 pa_sink_unref(u
->sink
);
2129 pa_source_unref(u
->source
);
2133 pa_rtpoll_free(u
->rtpoll
);
2136 pa_pstream_unlink(u
->pstream
);
2137 pa_pstream_unref(u
->pstream
);
2141 pa_pdispatch_unref(u
->pdispatch
);
2144 pa_socket_client_unref(u
->client
);
2147 pa_auth_cookie_unref(u
->auth_cookie
);
2150 pa_smoother_free(u
->smoother
);
2153 u
->core
->mainloop
->time_free(u
->time_event
);
2157 pa_mcalign_free(u
->mcalign
);
2161 pa_xfree(u
->sink_name
);
2163 pa_xfree(u
->source_name
);
2165 pa_xfree(u
->server_name
);
2167 pa_xfree(u
->device_description
);
2168 pa_xfree(u
->server_fqdn
);
2169 pa_xfree(u
->user_name
);