4 This file is part of polypaudio.
6 polypaudio is free software; you can redistribute it and/or modify
7 it under the terms of the GNU Lesser General Public License as published
8 by the Free Software Foundation; either version 2 of the License,
9 or (at your option) any later version.
11 polypaudio is distributed in the hope that it will be useful, but
12 WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 General Public License for more details.
16 You should have received a copy of the GNU Lesser General Public License
17 along with polypaudio; if not, write to the Free Software
18 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
30 #include <sys/types.h>
34 #include <polyp/version.h>
36 #include <polypcore/module.h>
37 #include <polypcore/util.h>
38 #include <polypcore/modargs.h>
39 #include <polypcore/log.h>
40 #include <polypcore/core-subscribe.h>
41 #include <polypcore/xmalloc.h>
42 #include <polypcore/sink-input.h>
43 #include <polypcore/pdispatch.h>
44 #include <polypcore/pstream.h>
45 #include <polypcore/pstream-util.h>
46 #include <polypcore/authkey.h>
47 #include <polypcore/socket-client.h>
48 #include <polypcore/socket-util.h>
49 #include <polypcore/authkey-prop.h>
52 #include "module-tunnel-sink-symdef.h"
53 PA_MODULE_DESCRIPTION("Tunnel module for sinks")
54 PA_MODULE_USAGE("server=<address> sink=<remote sink name> cookie=<filename> format=<sample format> channels=<number of channels> rate=<sample rate> sink_name=<name for the local sink>")
56 #include "module-tunnel-source-symdef.h"
57 PA_MODULE_DESCRIPTION("Tunnel module for sources")
58 PA_MODULE_USAGE("server=<address> source=<remote source name> cookie=<filename> format=<sample format> channels=<number of channels> rate=<sample rate> source_name=<name for the local source>")
61 PA_MODULE_AUTHOR("Lennart Poettering")
62 PA_MODULE_VERSION(PACKAGE_VERSION
)
64 #define DEFAULT_SINK_NAME "tunnel"
65 #define DEFAULT_SOURCE_NAME "tunnel"
67 #define DEFAULT_TLENGTH (44100*2*2/10) //(10240*8)
68 #define DEFAULT_MAXLENGTH ((DEFAULT_TLENGTH*3)/2)
69 #define DEFAULT_MINREQ 512
70 #define DEFAULT_PREBUF (DEFAULT_TLENGTH-DEFAULT_MINREQ)
71 #define DEFAULT_FRAGSIZE 1024
73 #define DEFAULT_TIMEOUT 5
75 #define LATENCY_INTERVAL 10
77 static const char* const valid_modargs
[] = {
93 static void command_stream_killed(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
96 static void command_request(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
);
99 static const pa_pdispatch_cb_t command_table
[PA_COMMAND_MAX
] = {
101 [PA_COMMAND_REQUEST
] = command_request
,
103 [PA_COMMAND_PLAYBACK_STREAM_KILLED
] = command_stream_killed
,
104 [PA_COMMAND_RECORD_STREAM_KILLED
] = command_stream_killed
108 pa_socket_client
*client
;
110 pa_pdispatch
*pdispatch
;
116 uint32_t requested_bytes
;
125 uint8_t auth_cookie
[PA_NATIVE_COOKIE_LENGTH
];
129 uint32_t device_index
;
132 pa_usec_t host_latency
;
134 pa_time_event
*time_event
;
136 int auth_cookie_in_property
;
139 static void close_stuff(struct userdata
*u
) {
143 pa_pstream_close(u
->pstream
);
144 pa_pstream_unref(u
->pstream
);
149 pa_pdispatch_unref(u
->pdispatch
);
154 pa_socket_client_unref(u
->client
);
160 pa_sink_disconnect(u
->sink
);
161 pa_sink_unref(u
->sink
);
166 pa_source_disconnect(u
->source
);
167 pa_source_unref(u
->source
);
173 u
->core
->mainloop
->time_free(u
->time_event
);
174 u
->time_event
= NULL
;
178 static void die(struct userdata
*u
) {
181 pa_module_unload_request(u
->module
);
184 static void command_stream_killed(pa_pdispatch
*pd
, PA_GCC_UNUSED
uint32_t command
, PA_GCC_UNUSED
uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
185 struct userdata
*u
= userdata
;
186 assert(pd
&& t
&& u
&& u
->pdispatch
== pd
);
188 pa_log(__FILE__
": stream killed");
193 static void send_prebuf_request(struct userdata
*u
) {
196 t
= pa_tagstruct_new(NULL
, 0);
197 pa_tagstruct_putu32(t
, PA_COMMAND_PREBUF_PLAYBACK_STREAM
);
198 pa_tagstruct_putu32(t
, u
->ctag
++);
199 pa_tagstruct_putu32(t
, u
->channel
);
200 pa_pstream_send_tagstruct(u
->pstream
, t
);
203 static void send_bytes(struct userdata
*u
) {
209 while (u
->requested_bytes
> 0) {
211 if (pa_sink_render(u
->sink
, u
->requested_bytes
, &chunk
) < 0) {
214 if (u
->requested_bytes
>= DEFAULT_TLENGTH
-DEFAULT_PREBUF
)
215 send_prebuf_request(u
);
220 pa_pstream_send_memblock(u
->pstream
, u
->channel
, 0, PA_SEEK_RELATIVE
, &chunk
);
221 pa_memblock_unref(chunk
.memblock
);
223 if (chunk
.length
> u
->requested_bytes
)
224 u
->requested_bytes
= 0;
226 u
->requested_bytes
-= chunk
.length
;
230 static void command_request(pa_pdispatch
*pd
, uint32_t command
, PA_GCC_UNUSED
uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
231 struct userdata
*u
= userdata
;
232 uint32_t bytes
, channel
;
233 assert(pd
&& command
== PA_COMMAND_REQUEST
&& t
&& u
&& u
->pdispatch
== pd
);
235 if (pa_tagstruct_getu32(t
, &channel
) < 0 ||
236 pa_tagstruct_getu32(t
, &bytes
) < 0 ||
237 !pa_tagstruct_eof(t
)) {
238 pa_log(__FILE__
": invalid protocol reply");
243 if (channel
!= u
->channel
) {
244 pa_log(__FILE__
": recieved data for invalid channel");
249 u
->requested_bytes
+= bytes
;
255 static void stream_get_latency_callback(pa_pdispatch
*pd
, uint32_t command
, PA_GCC_UNUSED
uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
256 struct userdata
*u
= userdata
;
257 pa_usec_t buffer_usec
, sink_usec
, source_usec
, transport_usec
;
259 uint32_t queue_length
;
260 int64_t write_index
, read_index
;
261 struct timeval local
, remote
, now
;
264 if (command
!= PA_COMMAND_REPLY
) {
265 if (command
== PA_COMMAND_ERROR
)
266 pa_log(__FILE__
": failed to get latency.");
268 pa_log(__FILE__
": protocol error.");
273 if (pa_tagstruct_get_usec(t
, &buffer_usec
) < 0 ||
274 pa_tagstruct_get_usec(t
, &sink_usec
) < 0 ||
275 pa_tagstruct_get_usec(t
, &source_usec
) < 0 ||
276 pa_tagstruct_get_boolean(t
, &playing
) < 0 ||
277 pa_tagstruct_getu32(t
, &queue_length
) < 0 ||
278 pa_tagstruct_get_timeval(t
, &local
) < 0 ||
279 pa_tagstruct_get_timeval(t
, &remote
) < 0 ||
280 pa_tagstruct_gets64(t
, &write_index
) < 0 ||
281 pa_tagstruct_gets64(t
, &read_index
) < 0 ||
282 !pa_tagstruct_eof(t
)) {
283 pa_log(__FILE__
": invalid reply.");
288 pa_gettimeofday(&now
);
290 if (pa_timeval_cmp(&local
, &remote
) < 0 && pa_timeval_cmp(&remote
, &now
)) {
291 /* local and remote seem to have synchronized clocks */
293 transport_usec
= pa_timeval_diff(&remote
, &local
);
295 transport_usec
= pa_timeval_diff(&now
, &remote
);
298 transport_usec
= pa_timeval_diff(&now
, &local
)/2;
301 u
->host_latency
= sink_usec
+ transport_usec
;
303 u
->host_latency
= source_usec
+ transport_usec
;
304 if (u
->host_latency
> sink_usec
)
305 u
->host_latency
-= sink_usec
;
310 /* pa_log(__FILE__": estimated host latency: %0.0f usec", (double) u->host_latency); */
313 static void request_latency(struct userdata
*u
) {
319 t
= pa_tagstruct_new(NULL
, 0);
321 pa_tagstruct_putu32(t
, PA_COMMAND_GET_PLAYBACK_LATENCY
);
323 pa_tagstruct_putu32(t
, PA_COMMAND_GET_RECORD_LATENCY
);
325 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
326 pa_tagstruct_putu32(t
, u
->channel
);
328 pa_gettimeofday(&now
);
329 pa_tagstruct_put_timeval(t
, &now
);
331 pa_pstream_send_tagstruct(u
->pstream
, t
);
332 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, stream_get_latency_callback
, u
);
335 static void create_stream_callback(pa_pdispatch
*pd
, uint32_t command
, PA_GCC_UNUSED
uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
336 struct userdata
*u
= userdata
;
337 assert(pd
&& u
&& u
->pdispatch
== pd
);
339 if (command
!= PA_COMMAND_REPLY
) {
340 if (command
== PA_COMMAND_ERROR
)
341 pa_log(__FILE__
": failed to create stream.");
343 pa_log(__FILE__
": protocol error.");
348 if (pa_tagstruct_getu32(t
, &u
->channel
) < 0 ||
349 pa_tagstruct_getu32(t
, &u
->device_index
) < 0 ||
351 pa_tagstruct_getu32(t
, &u
->requested_bytes
) < 0 ||
353 !pa_tagstruct_eof(t
)) {
354 pa_log(__FILE__
": invalid reply.");
365 static void setup_complete_callback(pa_pdispatch
*pd
, uint32_t command
, uint32_t tag
, pa_tagstruct
*t
, void *userdata
) {
366 struct userdata
*u
= userdata
;
368 char name
[256], un
[128], hn
[128];
372 assert(pd
&& u
&& u
->pdispatch
== pd
);
374 if (command
!= PA_COMMAND_REPLY
||
375 pa_tagstruct_getu32(t
, &u
->version
) < 0 ||
376 !pa_tagstruct_eof(t
)) {
377 if (command
== PA_COMMAND_ERROR
)
378 pa_log(__FILE__
": failed to authenticate");
380 pa_log(__FILE__
": protocol error.");
385 /* Minimum supported protocol version */
386 if (u
->version
< 8) {
387 pa_log(__FILE__
": incompatible protocol version");
393 snprintf(name
, sizeof(name
), "Tunnel from host '%s', user '%s', sink '%s'",
394 pa_get_host_name(hn
, sizeof(hn
)),
395 pa_get_user_name(un
, sizeof(un
)),
398 snprintf(name
, sizeof(name
), "Tunnel from host '%s', user '%s', source '%s'",
399 pa_get_host_name(hn
, sizeof(hn
)),
400 pa_get_user_name(un
, sizeof(un
)),
404 reply
= pa_tagstruct_new(NULL
, 0);
405 pa_tagstruct_putu32(reply
, PA_COMMAND_SET_CLIENT_NAME
);
406 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
407 pa_tagstruct_puts(reply
, name
);
408 pa_pstream_send_tagstruct(u
->pstream
, reply
);
409 /* We ignore the server's reply here */
411 reply
= pa_tagstruct_new(NULL
, 0);
413 pa_tagstruct_putu32(reply
, PA_COMMAND_CREATE_PLAYBACK_STREAM
);
414 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
415 pa_tagstruct_puts(reply
, name
);
416 pa_tagstruct_put_sample_spec(reply
, &u
->sink
->sample_spec
);
417 pa_tagstruct_put_channel_map(reply
, &u
->sink
->channel_map
);
418 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
);
419 pa_tagstruct_puts(reply
, u
->sink_name
);
420 pa_tagstruct_putu32(reply
, DEFAULT_MAXLENGTH
);
421 pa_tagstruct_put_boolean(reply
, 0);
422 pa_tagstruct_putu32(reply
, DEFAULT_TLENGTH
);
423 pa_tagstruct_putu32(reply
, DEFAULT_PREBUF
);
424 pa_tagstruct_putu32(reply
, DEFAULT_MINREQ
);
425 pa_tagstruct_putu32(reply
, 0);
426 pa_cvolume_reset(&volume
, u
->sink
->sample_spec
.channels
);
427 pa_tagstruct_put_cvolume(reply
, &volume
);
429 pa_tagstruct_putu32(reply
, PA_COMMAND_CREATE_RECORD_STREAM
);
430 pa_tagstruct_putu32(reply
, tag
= u
->ctag
++);
431 pa_tagstruct_puts(reply
, name
);
432 pa_tagstruct_put_sample_spec(reply
, &u
->source
->sample_spec
);
433 pa_tagstruct_put_channel_map(reply
, &u
->source
->channel_map
);
434 pa_tagstruct_putu32(reply
, PA_INVALID_INDEX
);
435 pa_tagstruct_puts(reply
, u
->source_name
);
436 pa_tagstruct_putu32(reply
, DEFAULT_MAXLENGTH
);
437 pa_tagstruct_put_boolean(reply
, 0);
438 pa_tagstruct_putu32(reply
, DEFAULT_FRAGSIZE
);
441 pa_pstream_send_tagstruct(u
->pstream
, reply
);
442 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, create_stream_callback
, u
);
445 static void pstream_die_callback(pa_pstream
*p
, void *userdata
) {
446 struct userdata
*u
= userdata
;
449 pa_log(__FILE__
": stream died.");
454 static void pstream_packet_callback(pa_pstream
*p
, pa_packet
*packet
, const void*creds
, void *userdata
) {
455 struct userdata
*u
= userdata
;
456 assert(p
&& packet
&& u
);
458 if (pa_pdispatch_run(u
->pdispatch
, packet
, creds
, u
) < 0) {
459 pa_log(__FILE__
": invalid packet");
465 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
) {
466 struct userdata
*u
= userdata
;
467 assert(p
&& chunk
&& u
);
469 if (channel
!= u
->channel
) {
470 pa_log(__FILE__
": recieved memory block on bad channel.");
475 pa_source_post(u
->source
, chunk
);
479 static void on_connection(pa_socket_client
*sc
, pa_iochannel
*io
, void *userdata
) {
480 struct userdata
*u
= userdata
;
483 assert(sc
&& u
&& u
->client
== sc
);
485 pa_socket_client_unref(u
->client
);
489 pa_log(__FILE__
": connection failed.");
490 pa_module_unload_request(u
->module
);
494 u
->pstream
= pa_pstream_new(u
->core
->mainloop
, io
, u
->core
->memblock_stat
);
495 u
->pdispatch
= pa_pdispatch_new(u
->core
->mainloop
, command_table
, PA_COMMAND_MAX
);
497 pa_pstream_set_die_callback(u
->pstream
, pstream_die_callback
, u
);
498 pa_pstream_set_recieve_packet_callback(u
->pstream
, pstream_packet_callback
, u
);
500 pa_pstream_set_recieve_memblock_callback(u
->pstream
, pstream_memblock_callback
, u
);
503 t
= pa_tagstruct_new(NULL
, 0);
504 pa_tagstruct_putu32(t
, PA_COMMAND_AUTH
);
505 pa_tagstruct_putu32(t
, tag
= u
->ctag
++);
506 pa_tagstruct_putu32(t
, PA_PROTOCOL_VERSION
);
507 pa_tagstruct_put_arbitrary(t
, u
->auth_cookie
, sizeof(u
->auth_cookie
));
508 pa_pstream_send_tagstruct(u
->pstream
, t
);
509 pa_pdispatch_register_reply(u
->pdispatch
, tag
, DEFAULT_TIMEOUT
, setup_complete_callback
, u
);
514 static void sink_notify(pa_sink
*sink
) {
516 assert(sink
&& sink
->userdata
);
522 static pa_usec_t
sink_get_latency(pa_sink
*sink
) {
526 assert(sink
&& sink
->userdata
);
531 if (l
> u
->requested_bytes
) {
532 l
-= u
->requested_bytes
;
533 usec
+= pa_bytes_to_usec(l
, &u
->sink
->sample_spec
);
536 usec
+= u
->host_latency
;
541 static pa_usec_t
source_get_latency(pa_source
*source
) {
543 assert(source
&& source
->userdata
);
544 u
= source
->userdata
;
546 return u
->host_latency
;
550 static void timeout_callback(pa_mainloop_api
*m
, pa_time_event
*e
, PA_GCC_UNUSED
const struct timeval
*tv
, void *userdata
) {
551 struct userdata
*u
= userdata
;
557 pa_gettimeofday(&ntv
);
558 ntv
.tv_sec
+= LATENCY_INTERVAL
;
559 m
->time_restart(e
, &ntv
);
562 static int load_key(struct userdata
*u
, const char*fn
) {
565 u
->auth_cookie_in_property
= 0;
567 if (!fn
&& pa_authkey_prop_get(u
->core
, PA_NATIVE_COOKIE_PROPERTY_NAME
, u
->auth_cookie
, sizeof(u
->auth_cookie
)) >= 0) {
568 pa_log_debug(__FILE__
": using already loaded auth cookie.");
569 pa_authkey_prop_ref(u
->core
, PA_NATIVE_COOKIE_PROPERTY_NAME
);
570 u
->auth_cookie_in_property
= 1;
575 fn
= PA_NATIVE_COOKIE_FILE
;
577 if (pa_authkey_load_auto(fn
, u
->auth_cookie
, sizeof(u
->auth_cookie
)) < 0)
580 pa_log_debug(__FILE__
": loading cookie from disk.");
582 if (pa_authkey_prop_put(u
->core
, PA_NATIVE_COOKIE_PROPERTY_NAME
, u
->auth_cookie
, sizeof(u
->auth_cookie
)) >= 0)
583 u
->auth_cookie_in_property
= 1;
588 int pa__init(pa_core
*c
, pa_module
*m
) {
589 pa_modargs
*ma
= NULL
;
590 struct userdata
*u
= NULL
;
595 if (!(ma
= pa_modargs_new(m
->argument
, valid_modargs
))) {
596 pa_log(__FILE__
": failed to parse module arguments");
600 u
= pa_xmalloc(sizeof(struct userdata
));
607 u
->server_name
= NULL
;
609 u
->sink_name
= pa_xstrdup(pa_modargs_get_value(ma
, "sink", NULL
));;
611 u
->requested_bytes
= 0;
613 u
->source_name
= pa_xstrdup(pa_modargs_get_value(ma
, "source", NULL
));;
617 u
->device_index
= u
->channel
= PA_INVALID_INDEX
;
619 u
->auth_cookie_in_property
= 0;
620 u
->time_event
= NULL
;
622 if (load_key(u
, pa_modargs_get_value(ma
, "cookie", NULL
)) < 0)
625 if (!(u
->server_name
= pa_xstrdup(pa_modargs_get_value(ma
, "server", NULL
)))) {
626 pa_log(__FILE__
": no server specified.");
630 ss
= c
->default_sample_spec
;
631 if (pa_modargs_get_sample_spec(ma
, &ss
) < 0) {
632 pa_log(__FILE__
": invalid sample format specification");
636 if (!(u
->client
= pa_socket_client_new_string(c
->mainloop
, u
->server_name
, PA_NATIVE_DEFAULT_PORT
))) {
637 pa_log(__FILE__
": failed to connect to server '%s'", u
->server_name
);
644 pa_socket_client_set_callback(u
->client
, on_connection
, u
);
647 if (!(u
->sink
= pa_sink_new(c
, __FILE__
, pa_modargs_get_value(ma
, "sink_name", DEFAULT_SINK_NAME
), 0, &ss
, NULL
))) {
648 pa_log(__FILE__
": failed to create sink.");
652 u
->sink
->notify
= sink_notify
;
653 u
->sink
->get_latency
= sink_get_latency
;
654 u
->sink
->userdata
= u
;
655 u
->sink
->description
= pa_sprintf_malloc("Tunnel to '%s%s%s'", u
->sink_name
? u
->sink_name
: "", u
->sink_name
? "@" : "", u
->server_name
);
657 pa_sink_set_owner(u
->sink
, m
);
659 if (!(u
->source
= pa_source_new(c
, __FILE__
, pa_modargs_get_value(ma
, "source_name", DEFAULT_SOURCE_NAME
), 0, &ss
, NULL
))) {
660 pa_log(__FILE__
": failed to create source.");
664 u
->source
->get_latency
= source_get_latency
;
665 u
->source
->userdata
= u
;
666 u
->source
->description
= pa_sprintf_malloc("Tunnel to '%s%s%s'", u
->source_name
? u
->source_name
: "", u
->source_name
? "@" : "", u
->server_name
);
668 pa_source_set_owner(u
->source
, m
);
671 pa_gettimeofday(&ntv
);
672 ntv
.tv_sec
+= LATENCY_INTERVAL
;
673 u
->time_event
= c
->mainloop
->time_new(c
->mainloop
, &ntv
, timeout_callback
, u
);
687 void pa__done(pa_core
*c
, pa_module
*m
) {
691 if (!(u
= m
->userdata
))
696 if (u
->auth_cookie_in_property
)
697 pa_authkey_prop_unref(c
, PA_NATIVE_COOKIE_PROPERTY_NAME
);
700 pa_xfree(u
->sink_name
);
702 pa_xfree(u
->source_name
);
704 pa_xfree(u
->server_name
);