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 otherwide 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_SINK_IS_OPENED(u->sink->thread_info.state))
702 if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
703 pa_sink_process_rewind(u->sink, 0);
706 if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
714 /* If this was no regular exit from the loop we have to continue
715 * processing messages until we received PA_MESSAGE_SHUTDOWN */
716 pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
717 pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
720 pa_log_debug("Thread shutting down");
724 /* Called from main context */
725 static void command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
726 struct userdata *u = userdata;
727 uint32_t bytes, channel;
730 pa_assert(command == PA_COMMAND_REQUEST);
733 pa_assert(u->pdispatch == pd);
735 if (pa_tagstruct_getu32(t, &channel) < 0 ||
736 pa_tagstruct_getu32(t, &bytes) < 0) {
737 pa_log("Invalid protocol reply");
741 if (channel != u->channel) {
742 pa_log("Received data for invalid channel");
746 pa_asyncmsgq_post(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REQUEST, NULL, bytes, NULL, NULL);
750 pa_module_unload_request(u->module, TRUE);
755 /* Called from main context */
756 static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
757 struct userdata *u = userdata;
758 pa_usec_t sink_usec, source_usec;
760 int64_t write_index, read_index;
761 struct timeval local, remote, now;
768 if (command != PA_COMMAND_REPLY) {
769 if (command == PA_COMMAND_ERROR)
770 pa_log("Failed to get latency.");
772 pa_log("Protocol error.");
776 if (pa_tagstruct_get_usec(t, &sink_usec) < 0 ||
777 pa_tagstruct_get_usec(t, &source_usec) < 0 ||
778 pa_tagstruct_get_boolean(t, &playing) < 0 ||
779 pa_tagstruct_get_timeval(t, &local) < 0 ||
780 pa_tagstruct_get_timeval(t, &remote) < 0 ||
781 pa_tagstruct_gets64(t, &write_index) < 0 ||
782 pa_tagstruct_gets64(t, &read_index) < 0) {
783 pa_log("Invalid reply.");
788 if (u->version >= 13) {
789 uint64_t underrun_for = 0, playing_for = 0;
791 if (pa_tagstruct_getu64(t, &underrun_for) < 0 ||
792 pa_tagstruct_getu64(t, &playing_for) < 0) {
793 pa_log("Invalid reply.");
799 if (!pa_tagstruct_eof(t)) {
800 pa_log("Invalid reply.");
804 if (tag < u->ignore_latency_before) {
808 pa_gettimeofday(&now);
810 /* Calculate transport usec */
811 if (pa_timeval_cmp(&local, &remote) < 0 && pa_timeval_cmp(&remote, &now)) {
812 /* local and remote seem to have synchronized clocks */
814 u->transport_usec = pa_timeval_diff(&remote, &local);
816 u->transport_usec = pa_timeval_diff(&now, &remote);
819 u->transport_usec = pa_timeval_diff(&now, &local)/2;
821 /* First, take the device's delay */
823 delay = (int64_t) sink_usec;
824 ss = &u->sink->sample_spec;
826 delay = (int64_t) source_usec;
827 ss = &u->source->sample_spec;
830 /* Add the length of our server-side buffer */
831 if (write_index >= read_index)
832 delay += (int64_t) pa_bytes_to_usec((uint64_t) (write_index-read_index), ss);
834 delay -= (int64_t) pa_bytes_to_usec((uint64_t) (read_index-write_index), ss);
836 /* Our measurements are already out of date, hence correct by the *
837 * transport latency */
839 delay -= (int64_t) u->transport_usec;
841 delay += (int64_t) u->transport_usec;
844 /* Now correct by what we have have read/written since we requested the update */
846 delay += (int64_t) pa_bytes_to_usec((uint64_t) u->counter_delta, ss);
848 delay -= (int64_t) pa_bytes_to_usec((uint64_t) u->counter_delta, ss);
852 pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UPDATE_LATENCY, 0, delay, NULL);
854 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_UPDATE_LATENCY, 0, delay, NULL);
861 pa_module_unload_request(u->module, TRUE);
864 /* Called from main context */
865 static void request_latency(struct userdata *u) {
871 t = pa_tagstruct_new(NULL, 0);
873 pa_tagstruct_putu32(t, PA_COMMAND_GET_PLAYBACK_LATENCY);
875 pa_tagstruct_putu32(t, PA_COMMAND_GET_RECORD_LATENCY);
877 pa_tagstruct_putu32(t, tag = u->ctag++);
878 pa_tagstruct_putu32(t, u->channel);
880 pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
882 pa_pstream_send_tagstruct(u->pstream, t);
883 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, u, NULL);
885 u->ignore_latency_before = tag;
886 u->counter_delta = 0;
889 /* Called from main context */
890 static void timeout_callback(pa_mainloop_api *m, pa_time_event *e, const struct timeval *t, void *userdata) {
891 struct userdata *u = userdata;
899 pa_core_rttime_restart(u->core, e, pa_rtclock_now() + LATENCY_INTERVAL);
902 /* Called from main context */
903 static void update_description(struct userdata *u) {
905 char un[128], hn[128];
910 if (!u->server_fqdn || !u->user_name || !u->device_description)
913 d = pa_sprintf_malloc("%s on %s@%s", u->device_description, u->user_name, u->server_fqdn);
916 pa_sink_set_description(u->sink, d);
917 pa_proplist_sets(u->sink->proplist, "tunnel.remote.user", u->user_name);
918 pa_proplist_sets(u->sink->proplist, "tunnel.remote.fqdn", u->server_fqdn);
919 pa_proplist_sets(u->sink->proplist, "tunnel.remote.description", u->device_description);
921 pa_source_set_description(u->source, d);
922 pa_proplist_sets(u->source->proplist, "tunnel.remote.user", u->user_name);
923 pa_proplist_sets(u->source->proplist, "tunnel.remote.fqdn", u->server_fqdn);
924 pa_proplist_sets(u->source->proplist, "tunnel.remote.description", u->device_description);
929 d = pa_sprintf_malloc("%s for %s@%s", u->device_description,
930 pa_get_user_name(un, sizeof(un)),
931 pa_get_host_name(hn, sizeof(hn)));
933 t = pa_tagstruct_new(NULL, 0);
935 pa_tagstruct_putu32(t, PA_COMMAND_SET_PLAYBACK_STREAM_NAME);
937 pa_tagstruct_putu32(t, PA_COMMAND_SET_RECORD_STREAM_NAME);
939 pa_tagstruct_putu32(t, u->ctag++);
940 pa_tagstruct_putu32(t, u->channel);
941 pa_tagstruct_puts(t, d);
942 pa_pstream_send_tagstruct(u->pstream, t);
947 /* Called from main context */
948 static void server_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
949 struct userdata *u = userdata;
952 const char *server_name, *server_version, *user_name, *host_name, *default_sink_name, *default_source_name;
958 if (command != PA_COMMAND_REPLY) {
959 if (command == PA_COMMAND_ERROR)
960 pa_log("Failed to get info.");
962 pa_log("Protocol error.");
966 if (pa_tagstruct_gets(t, &server_name) < 0 ||
967 pa_tagstruct_gets(t, &server_version) < 0 ||
968 pa_tagstruct_gets(t, &user_name) < 0 ||
969 pa_tagstruct_gets(t, &host_name) < 0 ||
970 pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
971 pa_tagstruct_gets(t, &default_sink_name) < 0 ||
972 pa_tagstruct_gets(t, &default_source_name) < 0 ||
973 pa_tagstruct_getu32(t, &cookie) < 0 ||
974 (u->version >= 15 && pa_tagstruct_get_channel_map(t, &cm) < 0)) {
976 pa_log("Parse failure");
980 if (!pa_tagstruct_eof(t)) {
981 pa_log("Packet too long");
985 pa_xfree(u->server_fqdn);
986 u->server_fqdn = pa_xstrdup(host_name);
988 pa_xfree(u->user_name);
989 u->user_name = pa_xstrdup(user_name);
991 update_description(u);
996 pa_module_unload_request(u->module, TRUE);
1001 /* Called from main context */
1002 static void sink_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1003 struct userdata *u = userdata;
1004 uint32_t idx, owner_module, monitor_source, flags;
1005 const char *name, *description, *monitor_source_name, *driver;
1016 pl = pa_proplist_new();
1018 if (command != PA_COMMAND_REPLY) {
1019 if (command == PA_COMMAND_ERROR)
1020 pa_log("Failed to get info.");
1022 pa_log("Protocol error.");
1026 if (pa_tagstruct_getu32(t, &idx) < 0 ||
1027 pa_tagstruct_gets(t, &name) < 0 ||
1028 pa_tagstruct_gets(t, &description) < 0 ||
1029 pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1030 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1031 pa_tagstruct_getu32(t, &owner_module) < 0 ||
1032 pa_tagstruct_get_cvolume(t, &volume) < 0 ||
1033 pa_tagstruct_get_boolean(t, &mute) < 0 ||
1034 pa_tagstruct_getu32(t, &monitor_source) < 0 ||
1035 pa_tagstruct_gets(t, &monitor_source_name) < 0 ||
1036 pa_tagstruct_get_usec(t, &latency) < 0 ||
1037 pa_tagstruct_gets(t, &driver) < 0 ||
1038 pa_tagstruct_getu32(t, &flags) < 0) {
1040 pa_log("Parse failure");
1044 if (u->version >= 13) {
1045 pa_usec_t configured_latency;
1047 if (pa_tagstruct_get_proplist(t, pl) < 0 ||
1048 pa_tagstruct_get_usec(t, &configured_latency) < 0) {
1050 pa_log("Parse failure");
1055 if (u->version >= 15) {
1056 pa_volume_t base_volume;
1057 uint32_t state, n_volume_steps, card;
1059 if (pa_tagstruct_get_volume(t, &base_volume) < 0 ||
1060 pa_tagstruct_getu32(t, &state) < 0 ||
1061 pa_tagstruct_getu32(t, &n_volume_steps) < 0 ||
1062 pa_tagstruct_getu32(t, &card) < 0) {
1064 pa_log("Parse failure");
1069 if (u->version >= 16) {
1073 if (pa_tagstruct_getu32(t, &n_ports)) {
1074 pa_log("Parse failure");
1078 for (uint32_t j = 0; j < n_ports; j++) {
1081 if (pa_tagstruct_gets(t, &s) < 0 || /* name */
1082 pa_tagstruct_gets(t, &s) < 0 || /* description */
1083 pa_tagstruct_getu32(t, &priority) < 0) {
1085 pa_log("Parse failure");
1090 if (pa_tagstruct_gets(t, &s) < 0) { /* active port */
1091 pa_log("Parse failure");
1096 if (u->version >= 21) {
1098 pa_format_info format;
1100 if (pa_tagstruct_getu8(t, &n_formats) < 0) { /* no. of formats */
1101 pa_log("Parse failure");
1105 for (uint8_t j = 0; j < n_formats; j++) {
1106 if (pa_tagstruct_get_format_info(t, &format)) { /* format info */
1107 pa_log("Parse failure");
1113 if (!pa_tagstruct_eof(t)) {
1114 pa_log("Packet too long");
1118 pa_proplist_free(pl);
1120 if (!u->sink_name || strcmp(name, u->sink_name))
1123 pa_xfree(u->device_description);
1124 u->device_description = pa_xstrdup(description);
1126 update_description(u);
1131 pa_module_unload_request(u->module, TRUE);
1132 pa_proplist_free(pl);
1135 /* Called from main context */
1136 static void sink_input_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1137 struct userdata *u = userdata;
1138 uint32_t idx, owner_module, client, sink;
1139 pa_usec_t buffer_usec, sink_usec;
1140 const char *name, *driver, *resample_method;
1141 pa_bool_t mute = FALSE;
1142 pa_sample_spec sample_spec;
1143 pa_channel_map channel_map;
1151 pl = pa_proplist_new();
1153 if (command != PA_COMMAND_REPLY) {
1154 if (command == PA_COMMAND_ERROR)
1155 pa_log("Failed to get info.");
1157 pa_log("Protocol error.");
1161 if (pa_tagstruct_getu32(t, &idx) < 0 ||
1162 pa_tagstruct_gets(t, &name) < 0 ||
1163 pa_tagstruct_getu32(t, &owner_module) < 0 ||
1164 pa_tagstruct_getu32(t, &client) < 0 ||
1165 pa_tagstruct_getu32(t, &sink) < 0 ||
1166 pa_tagstruct_get_sample_spec(t, &sample_spec) < 0 ||
1167 pa_tagstruct_get_channel_map(t, &channel_map) < 0 ||
1168 pa_tagstruct_get_cvolume(t, &volume) < 0 ||
1169 pa_tagstruct_get_usec(t, &buffer_usec) < 0 ||
1170 pa_tagstruct_get_usec(t, &sink_usec) < 0 ||
1171 pa_tagstruct_gets(t, &resample_method) < 0 ||
1172 pa_tagstruct_gets(t, &driver) < 0) {
1174 pa_log("Parse failure");
1178 if (u->version >= 11) {
1179 if (pa_tagstruct_get_boolean(t, &mute) < 0) {
1181 pa_log("Parse failure");
1186 if (u->version >= 13) {
1187 if (pa_tagstruct_get_proplist(t, pl) < 0) {
1189 pa_log("Parse failure");
1194 if (u->version >= 19) {
1195 if (pa_tagstruct_get_boolean(t, &b) < 0) {
1197 pa_log("Parse failure");
1202 if (u->version >= 20) {
1203 if (pa_tagstruct_get_boolean(t, &b) < 0 ||
1204 pa_tagstruct_get_boolean(t, &b) < 0) {
1206 pa_log("Parse failure");
1211 if (u->version >= 21) {
1212 pa_format_info format;
1214 if (pa_tagstruct_get_format_info(t, &format) < 0) {
1216 pa_log("Parse failure");
1221 if (!pa_tagstruct_eof(t)) {
1222 pa_log("Packet too long");
1226 pa_proplist_free(pl);
1228 if (idx != u->device_index)
1233 if ((u->version < 11 || !!mute == !!u->sink->muted) &&
1234 pa_cvolume_equal(&volume, &u->sink->real_volume))
1237 pa_sink_volume_changed(u->sink, &volume);
1239 if (u->version >= 11)
1240 pa_sink_mute_changed(u->sink, mute);
1245 pa_module_unload_request(u->module, TRUE);
1246 pa_proplist_free(pl);
1251 /* Called from main context */
1252 static void source_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1253 struct userdata *u = userdata;
1254 uint32_t idx, owner_module, monitor_of_sink, flags;
1255 const char *name, *description, *monitor_of_sink_name, *driver;
1260 pa_usec_t latency, configured_latency;
1266 pl = pa_proplist_new();
1268 if (command != PA_COMMAND_REPLY) {
1269 if (command == PA_COMMAND_ERROR)
1270 pa_log("Failed to get info.");
1272 pa_log("Protocol error.");
1276 if (pa_tagstruct_getu32(t, &idx) < 0 ||
1277 pa_tagstruct_gets(t, &name) < 0 ||
1278 pa_tagstruct_gets(t, &description) < 0 ||
1279 pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1280 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1281 pa_tagstruct_getu32(t, &owner_module) < 0 ||
1282 pa_tagstruct_get_cvolume(t, &volume) < 0 ||
1283 pa_tagstruct_get_boolean(t, &mute) < 0 ||
1284 pa_tagstruct_getu32(t, &monitor_of_sink) < 0 ||
1285 pa_tagstruct_gets(t, &monitor_of_sink_name) < 0 ||
1286 pa_tagstruct_get_usec(t, &latency) < 0 ||
1287 pa_tagstruct_gets(t, &driver) < 0 ||
1288 pa_tagstruct_getu32(t, &flags) < 0) {
1290 pa_log("Parse failure");
1294 if (u->version >= 13) {
1295 if (pa_tagstruct_get_proplist(t, pl) < 0 ||
1296 pa_tagstruct_get_usec(t, &configured_latency) < 0) {
1298 pa_log("Parse failure");
1303 if (u->version >= 15) {
1304 pa_volume_t base_volume;
1305 uint32_t state, n_volume_steps, card;
1307 if (pa_tagstruct_get_volume(t, &base_volume) < 0 ||
1308 pa_tagstruct_getu32(t, &state) < 0 ||
1309 pa_tagstruct_getu32(t, &n_volume_steps) < 0 ||
1310 pa_tagstruct_getu32(t, &card) < 0) {
1312 pa_log("Parse failure");
1317 if (u->version >= 16) {
1321 if (pa_tagstruct_getu32(t, &n_ports)) {
1322 pa_log("Parse failure");
1326 for (uint32_t j = 0; j < n_ports; j++) {
1329 if (pa_tagstruct_gets(t, &s) < 0 || /* name */
1330 pa_tagstruct_gets(t, &s) < 0 || /* description */
1331 pa_tagstruct_getu32(t, &priority) < 0) {
1333 pa_log("Parse failure");
1338 if (pa_tagstruct_gets(t, &s) < 0) { /* active port */
1339 pa_log("Parse failure");
1344 if (!pa_tagstruct_eof(t)) {
1345 pa_log("Packet too long");
1349 pa_proplist_free(pl);
1351 if (!u->source_name || strcmp(name, u->source_name))
1354 pa_xfree(u->device_description);
1355 u->device_description = pa_xstrdup(description);
1357 update_description(u);
1362 pa_module_unload_request(u->module, TRUE);
1363 pa_proplist_free(pl);
1368 /* Called from main context */
1369 static void request_info(struct userdata *u) {
1374 t = pa_tagstruct_new(NULL, 0);
1375 pa_tagstruct_putu32(t, PA_COMMAND_GET_SERVER_INFO);
1376 pa_tagstruct_putu32(t, tag = u->ctag++);
1377 pa_pstream_send_tagstruct(u->pstream, t);
1378 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, server_info_cb, u, NULL);
1381 t = pa_tagstruct_new(NULL, 0);
1382 pa_tagstruct_putu32(t, PA_COMMAND_GET_SINK_INPUT_INFO);
1383 pa_tagstruct_putu32(t, tag = u->ctag++);
1384 pa_tagstruct_putu32(t, u->device_index);
1385 pa_pstream_send_tagstruct(u->pstream, t);
1386 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, sink_input_info_cb, u, NULL);
1389 t = pa_tagstruct_new(NULL, 0);
1390 pa_tagstruct_putu32(t, PA_COMMAND_GET_SINK_INFO);
1391 pa_tagstruct_putu32(t, tag = u->ctag++);
1392 pa_tagstruct_putu32(t, PA_INVALID_INDEX);
1393 pa_tagstruct_puts(t, u->sink_name);
1394 pa_pstream_send_tagstruct(u->pstream, t);
1395 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, sink_info_cb, u, NULL);
1398 if (u->source_name) {
1399 t = pa_tagstruct_new(NULL, 0);
1400 pa_tagstruct_putu32(t, PA_COMMAND_GET_SOURCE_INFO);
1401 pa_tagstruct_putu32(t, tag = u->ctag++);
1402 pa_tagstruct_putu32(t, PA_INVALID_INDEX);
1403 pa_tagstruct_puts(t, u->source_name);
1404 pa_pstream_send_tagstruct(u->pstream, t);
1405 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, source_info_cb, u, NULL);
1410 /* Called from main context */
1411 static void command_subscribe_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1412 struct userdata *u = userdata;
1413 pa_subscription_event_type_t e;
1419 pa_assert(command == PA_COMMAND_SUBSCRIBE_EVENT);
1421 if (pa_tagstruct_getu32(t, &e) < 0 ||
1422 pa_tagstruct_getu32(t, &idx) < 0) {
1423 pa_log("Invalid protocol reply");
1424 pa_module_unload_request(u->module, TRUE);
1428 if (e != (PA_SUBSCRIPTION_EVENT_SERVER|PA_SUBSCRIPTION_EVENT_CHANGE) &&
1430 e != (PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE) &&
1431 e != (PA_SUBSCRIPTION_EVENT_SINK|PA_SUBSCRIPTION_EVENT_CHANGE)
1433 e != (PA_SUBSCRIPTION_EVENT_SOURCE|PA_SUBSCRIPTION_EVENT_CHANGE)
1441 /* Called from main context */
1442 static void start_subscribe(struct userdata *u) {
1446 t = pa_tagstruct_new(NULL, 0);
1447 pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE);
1448 pa_tagstruct_putu32(t, u->ctag++);
1449 pa_tagstruct_putu32(t, PA_SUBSCRIPTION_MASK_SERVER|
1451 PA_SUBSCRIPTION_MASK_SINK_INPUT|PA_SUBSCRIPTION_MASK_SINK
1453 PA_SUBSCRIPTION_MASK_SOURCE
1457 pa_pstream_send_tagstruct(u->pstream, t);
1460 /* Called from main context */
1461 static void create_stream_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1462 struct userdata *u = userdata;
1469 pa_assert(u->pdispatch == pd);
1471 if (command != PA_COMMAND_REPLY) {
1472 if (command == PA_COMMAND_ERROR)
1473 pa_log("Failed to create stream.");
1475 pa_log("Protocol error.");
1479 if (pa_tagstruct_getu32(t, &u->channel) < 0 ||
1480 pa_tagstruct_getu32(t, &u->device_index) < 0
1482 || pa_tagstruct_getu32(t, &bytes) < 0
1487 if (u->version >= 9) {
1489 if (pa_tagstruct_getu32(t, &u->maxlength) < 0 ||
1490 pa_tagstruct_getu32(t, &u->tlength) < 0 ||
1491 pa_tagstruct_getu32(t, &u->prebuf) < 0 ||
1492 pa_tagstruct_getu32(t, &u->minreq) < 0)
1495 if (pa_tagstruct_getu32(t, &u->maxlength) < 0 ||
1496 pa_tagstruct_getu32(t, &u->fragsize) < 0)
1501 if (u->version >= 12) {
1504 uint32_t device_index;
1506 pa_bool_t suspended;
1508 if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1509 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1510 pa_tagstruct_getu32(t, &device_index) < 0 ||
1511 pa_tagstruct_gets(t, &dn) < 0 ||
1512 pa_tagstruct_get_boolean(t, &suspended) < 0)
1516 pa_xfree(u->sink_name);
1517 u->sink_name = pa_xstrdup(dn);
1519 pa_xfree(u->source_name);
1520 u->source_name = pa_xstrdup(dn);
1524 if (u->version >= 13) {
1527 if (pa_tagstruct_get_usec(t, &usec) < 0)
1530 /* #ifdef TUNNEL_SINK */
1531 /* pa_sink_set_latency_range(u->sink, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1533 /* pa_source_set_latency_range(u->source, usec + MIN_NETWORK_LATENCY_USEC, 0); */
1537 if (u->version >= 21) {
1538 pa_format_info format;
1540 if (pa_tagstruct_get_format_info(t, &format) < 0)
1544 if (!pa_tagstruct_eof(t))
1550 pa_assert(!u->time_event);
1551 u->time_event = pa_core_rttime_new(u->core, pa_rtclock_now() + LATENCY_INTERVAL, timeout_callback, u);
1555 pa_log_debug("Stream created.");
1558 pa_asyncmsgq_post(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_REQUEST, NULL, bytes, NULL, NULL);
1564 pa_log("Invalid reply. (Create stream)");
1567 pa_module_unload_request(u->module, TRUE);
1571 /* Called from main context */
1572 static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1573 struct userdata *u = userdata;
1574 pa_tagstruct *reply;
1575 char name[256], un[128], hn[128];
1582 pa_assert(u->pdispatch == pd);
1584 if (command != PA_COMMAND_REPLY ||
1585 pa_tagstruct_getu32(t, &u->version) < 0 ||
1586 !pa_tagstruct_eof(t)) {
1588 if (command == PA_COMMAND_ERROR)
1589 pa_log("Failed to authenticate");
1591 pa_log("Protocol error.");
1596 /* Minimum supported protocol version */
1597 if (u->version < 8) {
1598 pa_log("Incompatible protocol version");
1602 /* Starting with protocol version 13 the MSB of the version tag
1603 reflects if shm is enabled for this connection or not. We don't
1604 support SHM here at all, so we just ignore this. */
1606 if (u->version >= 13)
1607 u->version &= 0x7FFFFFFFU;
1609 pa_log_debug("Protocol version: remote %u, local %u", u->version, PA_PROTOCOL_VERSION);
1612 pa_proplist_setf(u->sink->proplist, "tunnel.remote_version", "%u", u->version);
1613 pa_sink_update_proplist(u->sink, 0, NULL);
1615 pa_snprintf(name, sizeof(name), "%s for %s@%s",
1617 pa_get_user_name(un, sizeof(un)),
1618 pa_get_host_name(hn, sizeof(hn)));
1620 pa_proplist_setf(u->source->proplist, "tunnel.remote_version", "%u", u->version);
1621 pa_source_update_proplist(u->source, 0, NULL);
1623 pa_snprintf(name, sizeof(name), "%s for %s@%s",
1625 pa_get_user_name(un, sizeof(un)),
1626 pa_get_host_name(hn, sizeof(hn)));
1629 reply = pa_tagstruct_new(NULL, 0);
1630 pa_tagstruct_putu32(reply, PA_COMMAND_SET_CLIENT_NAME);
1631 pa_tagstruct_putu32(reply, u->ctag++);
1633 if (u->version >= 13) {
1635 pl = pa_proplist_new();
1636 pa_proplist_sets(pl, PA_PROP_APPLICATION_ID, "org.PulseAudio.PulseAudio");
1637 pa_proplist_sets(pl, PA_PROP_APPLICATION_VERSION, PACKAGE_VERSION);
1638 pa_init_proplist(pl);
1639 pa_tagstruct_put_proplist(reply, pl);
1640 pa_proplist_free(pl);
1642 pa_tagstruct_puts(reply, "PulseAudio");
1644 pa_pstream_send_tagstruct(u->pstream, reply);
1645 /* We ignore the server's reply here */
1647 reply = pa_tagstruct_new(NULL, 0);
1649 if (u->version < 13)
1650 /* Only for older PA versions we need to fill in the maxlength */
1651 u->maxlength = 4*1024*1024;
1654 u->tlength = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_TLENGTH_MSEC, &u->sink->sample_spec);
1655 u->minreq = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_MINREQ_MSEC, &u->sink->sample_spec);
1656 u->prebuf = u->tlength;
1658 u->fragsize = (uint32_t) pa_usec_to_bytes(PA_USEC_PER_MSEC * DEFAULT_FRAGSIZE_MSEC, &u->source->sample_spec);
1662 pa_tagstruct_putu32(reply, PA_COMMAND_CREATE_PLAYBACK_STREAM);
1663 pa_tagstruct_putu32(reply, tag = u->ctag++);
1665 if (u->version < 13)
1666 pa_tagstruct_puts(reply, name);
1668 pa_tagstruct_put_sample_spec(reply, &u->sink->sample_spec);
1669 pa_tagstruct_put_channel_map(reply, &u->sink->channel_map);
1670 pa_tagstruct_putu32(reply, PA_INVALID_INDEX);
1671 pa_tagstruct_puts(reply, u->sink_name);
1672 pa_tagstruct_putu32(reply, u->maxlength);
1673 pa_tagstruct_put_boolean(reply, !PA_SINK_IS_OPENED(pa_sink_get_state(u->sink)));
1674 pa_tagstruct_putu32(reply, u->tlength);
1675 pa_tagstruct_putu32(reply, u->prebuf);
1676 pa_tagstruct_putu32(reply, u->minreq);
1677 pa_tagstruct_putu32(reply, 0);
1678 pa_cvolume_reset(&volume, u->sink->sample_spec.channels);
1679 pa_tagstruct_put_cvolume(reply, &volume);
1681 pa_tagstruct_putu32(reply, PA_COMMAND_CREATE_RECORD_STREAM);
1682 pa_tagstruct_putu32(reply, tag = u->ctag++);
1684 if (u->version < 13)
1685 pa_tagstruct_puts(reply, name);
1687 pa_tagstruct_put_sample_spec(reply, &u->source->sample_spec);
1688 pa_tagstruct_put_channel_map(reply, &u->source->channel_map);
1689 pa_tagstruct_putu32(reply, PA_INVALID_INDEX);
1690 pa_tagstruct_puts(reply, u->source_name);
1691 pa_tagstruct_putu32(reply, u->maxlength);
1692 pa_tagstruct_put_boolean(reply, !PA_SOURCE_IS_OPENED(pa_source_get_state(u->source)));
1693 pa_tagstruct_putu32(reply, u->fragsize);
1696 if (u->version >= 12) {
1697 pa_tagstruct_put_boolean(reply, FALSE); /* no_remap */
1698 pa_tagstruct_put_boolean(reply, FALSE); /* no_remix */
1699 pa_tagstruct_put_boolean(reply, FALSE); /* fix_format */
1700 pa_tagstruct_put_boolean(reply, FALSE); /* fix_rate */
1701 pa_tagstruct_put_boolean(reply, FALSE); /* fix_channels */
1702 pa_tagstruct_put_boolean(reply, TRUE); /* no_move */
1703 pa_tagstruct_put_boolean(reply, FALSE); /* variable_rate */
1706 if (u->version >= 13) {
1709 pa_tagstruct_put_boolean(reply, FALSE); /* start muted/peak detect*/
1710 pa_tagstruct_put_boolean(reply, TRUE); /* adjust_latency */
1712 pl = pa_proplist_new();
1713 pa_proplist_sets(pl, PA_PROP_MEDIA_NAME, name);
1714 pa_proplist_sets(pl, PA_PROP_MEDIA_ROLE, "abstract");
1715 pa_tagstruct_put_proplist(reply, pl);
1716 pa_proplist_free(pl);
1719 pa_tagstruct_putu32(reply, PA_INVALID_INDEX); /* direct on input */
1723 if (u->version >= 14) {
1725 pa_tagstruct_put_boolean(reply, FALSE); /* volume_set */
1727 pa_tagstruct_put_boolean(reply, TRUE); /* early rquests */
1730 if (u->version >= 15) {
1732 pa_tagstruct_put_boolean(reply, FALSE); /* muted_set */
1734 pa_tagstruct_put_boolean(reply, FALSE); /* don't inhibit auto suspend */
1735 pa_tagstruct_put_boolean(reply, FALSE); /* fail on suspend */
1739 if (u->version >= 17)
1740 pa_tagstruct_put_boolean(reply, FALSE); /* relative volume */
1742 if (u->version >= 18)
1743 pa_tagstruct_put_boolean(reply, FALSE); /* passthrough stream */
1747 if (u->version >= 21) {
1748 /* We're not using the extended API, so n_formats = 0 and that's that */
1749 pa_tagstruct_putu8(t, 0);
1753 pa_pstream_send_tagstruct(u->pstream, reply);
1754 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, create_stream_callback, u, NULL);
1756 pa_log_debug("Connection authenticated, creating stream ...");
1761 pa_module_unload_request(u->module, TRUE);
1764 /* Called from main context */
1765 static void pstream_die_callback(pa_pstream *p, void *userdata) {
1766 struct userdata *u = userdata;
1771 pa_log_warn("Stream died.");
1772 pa_module_unload_request(u->module, TRUE);
1775 /* Called from main context */
1776 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
1777 struct userdata *u = userdata;
1783 if (pa_pdispatch_run(u->pdispatch, packet, creds, u) < 0) {
1784 pa_log("Invalid packet");
1785 pa_module_unload_request(u->module, TRUE);
1791 /* Called from main context */
1792 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) {
1793 struct userdata *u = userdata;
1799 if (channel != u->channel) {
1800 pa_log("Received memory block on bad channel.");
1801 pa_module_unload_request(u->module, TRUE);
1805 pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_POST, PA_UINT_TO_PTR(seek), offset, chunk);
1807 u->counter_delta += (int64_t) chunk->length;
1811 /* Called from main context */
1812 static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) {
1813 struct userdata *u = userdata;
1819 pa_assert(u->client == sc);
1821 pa_socket_client_unref(u->client);
1825 pa_log("Connection failed: %s", pa_cstrerror(errno));
1826 pa_module_unload_request(u->module, TRUE);
1830 u->pstream = pa_pstream_new(u->core->mainloop, io, u->core->mempool);
1831 u->pdispatch = pa_pdispatch_new(u->core->mainloop, TRUE, command_table, PA_COMMAND_MAX);
1833 pa_pstream_set_die_callback(u->pstream, pstream_die_callback, u);
1834 pa_pstream_set_recieve_packet_callback(u->pstream, pstream_packet_callback, u);
1836 pa_pstream_set_recieve_memblock_callback(u->pstream, pstream_memblock_callback, u);
1839 t = pa_tagstruct_new(NULL, 0);
1840 pa_tagstruct_putu32(t, PA_COMMAND_AUTH);
1841 pa_tagstruct_putu32(t, tag = u->ctag++);
1842 pa_tagstruct_putu32(t, PA_PROTOCOL_VERSION);
1844 pa_tagstruct_put_arbitrary(t, pa_auth_cookie_read(u->auth_cookie, PA_NATIVE_COOKIE_LENGTH), PA_NATIVE_COOKIE_LENGTH);
1850 if (pa_iochannel_creds_supported(io))
1851 pa_iochannel_creds_enable(io);
1853 ucred.uid = getuid();
1854 ucred.gid = getgid();
1856 pa_pstream_send_tagstruct_with_creds(u->pstream, t, &ucred);
1859 pa_pstream_send_tagstruct(u->pstream, t);
1862 pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, u, NULL);
1864 pa_log_debug("Connection established, authenticating ...");
1869 /* Called from main context */
1870 static void sink_set_volume(pa_sink *sink) {
1878 t = pa_tagstruct_new(NULL, 0);
1879 pa_tagstruct_putu32(t, PA_COMMAND_SET_SINK_INPUT_VOLUME);
1880 pa_tagstruct_putu32(t, u->ctag++);
1881 pa_tagstruct_putu32(t, u->device_index);
1882 pa_tagstruct_put_cvolume(t, &sink->real_volume);
1883 pa_pstream_send_tagstruct(u->pstream, t);
1886 /* Called from main context */
1887 static void sink_set_mute(pa_sink *sink) {
1895 if (u->version < 11)
1898 t = pa_tagstruct_new(NULL, 0);
1899 pa_tagstruct_putu32(t, PA_COMMAND_SET_SINK_INPUT_MUTE);
1900 pa_tagstruct_putu32(t, u->ctag++);
1901 pa_tagstruct_putu32(t, u->device_index);
1902 pa_tagstruct_put_boolean(t, !!sink->muted);
1903 pa_pstream_send_tagstruct(u->pstream, t);
1908 int pa__init(pa_module*m) {
1909 pa_modargs *ma = NULL;
1910 struct userdata *u = NULL;
1915 pa_sink_new_data data;
1917 pa_source_new_data data;
1922 if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
1923 pa_log("Failed to parse module arguments");
1927 m->userdata = u = pa_xnew0(struct userdata, 1);
1931 u->pdispatch = NULL;
1933 u->server_name = NULL;
1935 u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));;
1937 u->requested_bytes = 0;
1939 u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));;
1942 u->smoother = pa_smoother_new(
1951 u->device_index = u->channel = PA_INVALID_INDEX;
1952 u->time_event = NULL;
1953 u->ignore_latency_before = 0;
1954 u->transport_usec = u->thread_transport_usec = 0;
1955 u->remote_suspended = u->remote_corked = FALSE;
1956 u->counter = u->counter_delta = 0;
1958 u->rtpoll = pa_rtpoll_new();
1959 pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);
1961 if (!(u->auth_cookie = pa_auth_cookie_get(u->core, pa_modargs_get_value(ma, "cookie", PA_NATIVE_COOKIE_FILE), PA_NATIVE_COOKIE_LENGTH)))
1964 if (!(u->server_name = pa_xstrdup(pa_modargs_get_value(ma, "server", NULL)))) {
1965 pa_log("No server specified.");
1969 ss = m->core->default_sample_spec;
1970 map = m->core->default_channel_map;
1971 if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
1972 pa_log("Invalid sample format specification");
1976 if (!(u->client = pa_socket_client_new_string(m->core->mainloop, TRUE, u->server_name, PA_NATIVE_DEFAULT_PORT))) {
1977 pa_log("Failed to connect to server '%s'", u->server_name);
1981 pa_socket_client_set_callback(u->client, on_connection, u);
1985 if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL))))
1986 dn = pa_sprintf_malloc("tunnel-sink.%s", u->server_name);
1988 pa_sink_new_data_init(&data);
1989 data.driver = __FILE__;
1991 data.namereg_fail = TRUE;
1992 pa_sink_new_data_set_name(&data, dn);
1993 pa_sink_new_data_set_sample_spec(&data, &ss);
1994 pa_sink_new_data_set_channel_map(&data, &map);
1995 pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->sink_name), u->sink_name ? " on " : "", u->server_name);
1996 pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
1998 pa_proplist_sets(data.proplist, "tunnel.remote.sink", u->sink_name);
2000 if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
2001 pa_log("Invalid properties");
2002 pa_sink_new_data_done(&data);
2006 u->sink = pa_sink_new(m->core, &data, PA_SINK_NETWORK|PA_SINK_LATENCY|PA_SINK_HW_VOLUME_CTRL|PA_SINK_HW_MUTE_CTRL);
2007 pa_sink_new_data_done(&data);
2010 pa_log("Failed to create sink.");
2014 u->sink->parent.process_msg = sink_process_msg;
2015 u->sink->userdata = u;
2016 u->sink->set_state = sink_set_state;
2017 u->sink->set_volume = sink_set_volume;
2018 u->sink->set_mute = sink_set_mute;
2020 u->sink->refresh_volume = u->sink->refresh_muted = FALSE;
2022 /* pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
2024 pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
2025 pa_sink_set_rtpoll(u->sink, u->rtpoll);
2029 if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL))))
2030 dn = pa_sprintf_malloc("tunnel-source.%s", u->server_name);
2032 pa_source_new_data_init(&data);
2033 data.driver = __FILE__;
2035 data.namereg_fail = TRUE;
2036 pa_source_new_data_set_name(&data, dn);
2037 pa_source_new_data_set_sample_spec(&data, &ss);
2038 pa_source_new_data_set_channel_map(&data, &map);
2039 pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->source_name), u->source_name ? " on " : "", u->server_name);
2040 pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
2042 pa_proplist_sets(data.proplist, "tunnel.remote.source", u->source_name);
2044 if (pa_modargs_get_proplist(ma, "source_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
2045 pa_log("Invalid properties");
2046 pa_source_new_data_done(&data);
2050 u->source = pa_source_new(m->core, &data, PA_SOURCE_NETWORK|PA_SOURCE_LATENCY);
2051 pa_source_new_data_done(&data);
2054 pa_log("Failed to create source.");
2058 u->source->parent.process_msg = source_process_msg;
2059 u->source->set_state = source_set_state;
2060 u->source->userdata = u;
2062 /* pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
2064 pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
2065 pa_source_set_rtpoll(u->source, u->rtpoll);
2067 u->mcalign = pa_mcalign_new(pa_frame_size(&u->source->sample_spec));
2072 u->time_event = NULL;
2074 u->maxlength = (uint32_t) -1;
2076 u->tlength = u->minreq = u->prebuf = (uint32_t) -1;
2078 u->fragsize = (uint32_t) -1;
2081 if (!(u->thread = pa_thread_new("module-tunnel", thread_func, u))) {
2082 pa_log("Failed to create thread.");
2087 pa_sink_put(u->sink);
2089 pa_source_put(u->source);
2092 pa_modargs_free(ma);
2100 pa_modargs_free(ma);
2107 void pa__done(pa_module*m) {
2112 if (!(u = m->userdata))
2117 pa_sink_unlink(u->sink);
2120 pa_source_unlink(u->source);
2124 pa_asyncmsgq_send(u->thread_mq.inq, NULL, PA_MESSAGE_SHUTDOWN, NULL, 0, NULL);
2125 pa_thread_free(u->thread);
2128 pa_thread_mq_done(&u->thread_mq);
2132 pa_sink_unref(u->sink);
2135 pa_source_unref(u->source);
2139 pa_rtpoll_free(u->rtpoll);
2142 pa_pstream_unlink(u->pstream);
2143 pa_pstream_unref(u->pstream);
2147 pa_pdispatch_unref(u->pdispatch);
2150 pa_socket_client_unref(u->client);
2153 pa_auth_cookie_unref(u->auth_cookie);
2156 pa_smoother_free(u->smoother);
2159 u->core->mainloop->time_free(u->time_event);
2163 pa_mcalign_free(u->mcalign);
2167 pa_xfree(u->sink_name);
2169 pa_xfree(u->source_name);
2171 pa_xfree(u->server_name);
2173 pa_xfree(u->device_description);
2174 pa_xfree(u->server_fqdn);
2175 pa_xfree(u->user_name);