mirror of
https://gitlab.freedesktop.org/pulseaudio/pulseaudio.git
synced 2025-11-03 09:01:50 -05:00
raop: Add UDP protocol handling
There are two versions in the RAOP protocol; one uses TCP and the other uses UDP. Current raop implementation only supports TCP version. This patch adds an initial UDP protocol support for RAOP. It is based on Martin Blanchard's work (http://repo.or.cz/w/pulseaudio-raopUDP.git/shortlog/refs/heads/raop) which is inspired by Christophe Fergeau's work (https://github.com/zx2c4/pulseaudio-raop2). Matrin's modifications were edited by Hajime Fujita, so that it would support both TCP and UDP protocol in a single module. Also this patch includes a fix that was found thanks to Matthias, who reported that his ALAC codec support fixed the issue. https://bugs.freedesktop.org/show_bug.cgi?id=42804#c30
This commit is contained in:
Hajime Fujita
Tanu Kaskinen
parent
43bf134aa1
commit
1e51282621
3 changed files with 1400 additions and 159 deletions
|
|
@ -66,12 +66,13 @@ PA_MODULE_USAGE(
|
|||
"sink_name=<name for the sink> "
|
||||
"sink_properties=<properties for the sink> "
|
||||
"server=<address> "
|
||||
"protocol=<transport protocol> "
|
||||
"encryption=<encryption type> "
|
||||
"codec=<audio codec> "
|
||||
"format=<sample format> "
|
||||
"rate=<sample rate> "
|
||||
"channels=<number of channels>");
|
||||
|
||||
#define DEFAULT_SINK_NAME "raop"
|
||||
|
||||
struct userdata {
|
||||
pa_core *core;
|
||||
pa_module *module;
|
||||
|
|
@ -82,6 +83,8 @@ struct userdata {
|
|||
pa_rtpoll_item *rtpoll_item;
|
||||
pa_thread *thread;
|
||||
|
||||
pa_raop_protocol_t protocol;
|
||||
|
||||
pa_memchunk raw_memchunk;
|
||||
pa_memchunk encoded_memchunk;
|
||||
|
||||
|
|
@ -97,7 +100,6 @@ struct userdata {
|
|||
int32_t rate;
|
||||
|
||||
pa_smoother *smoother;
|
||||
int fd;
|
||||
|
||||
int64_t offset;
|
||||
int64_t encoding_overhead;
|
||||
|
|
@ -107,12 +109,26 @@ struct userdata {
|
|||
pa_raop_client *raop;
|
||||
|
||||
size_t block_size;
|
||||
|
||||
/* Members only for the TCP protocol */
|
||||
int tcp_fd;
|
||||
|
||||
/* Members only for the UDP protocol */
|
||||
int udp_control_fd;
|
||||
int udp_timing_fd;
|
||||
|
||||
/* For UDP thread wakeup clock calculation */
|
||||
pa_usec_t udp_playback_start;
|
||||
uint32_t udp_sent_packets;
|
||||
};
|
||||
|
||||
static const char* const valid_modargs[] = {
|
||||
"sink_name",
|
||||
"sink_properties",
|
||||
"server",
|
||||
"protocol",
|
||||
"encryption",
|
||||
"codec",
|
||||
"format",
|
||||
"rate",
|
||||
"channels",
|
||||
|
|
@ -120,23 +136,26 @@ static const char* const valid_modargs[] = {
|
|||
};
|
||||
|
||||
enum {
|
||||
SINK_MESSAGE_PASS_SOCKET = PA_SINK_MESSAGE_MAX,
|
||||
SINK_MESSAGE_RIP_SOCKET
|
||||
SINK_MESSAGE_TCP_PASS_SOCKET = PA_SINK_MESSAGE_MAX,
|
||||
SINK_MESSAGE_TCP_RIP_SOCKET,
|
||||
SINK_MESSAGE_UDP_SETUP,
|
||||
SINK_MESSAGE_UDP_RECORD,
|
||||
SINK_MESSAGE_UDP_DISCONNECTED,
|
||||
};
|
||||
|
||||
/* Forward declarations: */
|
||||
static void sink_set_volume_cb(pa_sink *);
|
||||
|
||||
static void on_connection(int fd, void *userdata) {
|
||||
static void tcp_on_connection(int fd, void *userdata) {
|
||||
int so_sndbuf = 0;
|
||||
socklen_t sl = sizeof(int);
|
||||
struct userdata *u = userdata;
|
||||
pa_assert(u);
|
||||
|
||||
pa_assert(u->fd < 0);
|
||||
u->fd = fd;
|
||||
pa_assert(u->tcp_fd < 0);
|
||||
u->tcp_fd = fd;
|
||||
|
||||
if (getsockopt(u->fd, SOL_SOCKET, SO_SNDBUF, &so_sndbuf, &sl) < 0)
|
||||
if (getsockopt(u->tcp_fd, SOL_SOCKET, SO_SNDBUF, &so_sndbuf, &sl) < 0)
|
||||
pa_log_warn("getsockopt(SO_SNDBUF) failed: %s", pa_cstrerror(errno));
|
||||
else {
|
||||
pa_log_debug("SO_SNDBUF is %zu.", (size_t) so_sndbuf);
|
||||
|
|
@ -148,19 +167,28 @@ static void on_connection(int fd, void *userdata) {
|
|||
|
||||
pa_log_debug("Connection authenticated, handing fd to IO thread...");
|
||||
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_PASS_SOCKET, NULL, 0, NULL, NULL);
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_TCP_PASS_SOCKET, NULL, 0, NULL, NULL);
|
||||
}
|
||||
|
||||
static void on_close(void*userdata) {
|
||||
static void tcp_on_close(void*userdata) {
|
||||
struct userdata *u = userdata;
|
||||
pa_assert(u);
|
||||
|
||||
pa_log_debug("Connection closed, informing IO thread...");
|
||||
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_RIP_SOCKET, NULL, 0, NULL, NULL);
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_TCP_RIP_SOCKET, NULL, 0, NULL, NULL);
|
||||
}
|
||||
|
||||
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
|
||||
static pa_usec_t sink_get_latency(const struct userdata *u) {
|
||||
pa_usec_t w, r;
|
||||
|
||||
r = pa_smoother_get(u->smoother, pa_rtclock_now());
|
||||
w = pa_bytes_to_usec((u->offset - u->encoding_overhead + (u->encoded_memchunk.length / u->encoding_ratio)), &u->sink->sample_spec);
|
||||
|
||||
return w > r ? w - r : 0;
|
||||
}
|
||||
|
||||
static int tcp_sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
|
||||
struct userdata *u = PA_SINK(o)->userdata;
|
||||
|
||||
switch (code) {
|
||||
|
|
@ -175,8 +203,8 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
|
|||
pa_smoother_pause(u->smoother, pa_rtclock_now());
|
||||
|
||||
/* Issue a FLUSH if we are connected. */
|
||||
if (u->fd >= 0) {
|
||||
pa_raop_flush(u->raop);
|
||||
if (u->tcp_fd >= 0) {
|
||||
pa_raop_client_flush(u->raop);
|
||||
}
|
||||
break;
|
||||
|
||||
|
|
@ -188,10 +216,10 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
|
|||
|
||||
/* The connection can be closed when idle, so check to
|
||||
* see if we need to reestablish it. */
|
||||
if (u->fd < 0)
|
||||
pa_raop_connect(u->raop);
|
||||
if (u->tcp_fd < 0)
|
||||
pa_raop_client_connect(u->raop);
|
||||
else
|
||||
pa_raop_flush(u->raop);
|
||||
pa_raop_client_flush(u->raop);
|
||||
}
|
||||
|
||||
break;
|
||||
|
|
@ -205,37 +233,32 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
|
|||
break;
|
||||
|
||||
case PA_SINK_MESSAGE_GET_LATENCY: {
|
||||
pa_usec_t w, r;
|
||||
|
||||
r = pa_smoother_get(u->smoother, pa_rtclock_now());
|
||||
w = pa_bytes_to_usec((u->offset - u->encoding_overhead + (u->encoded_memchunk.length / u->encoding_ratio)), &u->sink->sample_spec);
|
||||
|
||||
*((pa_usec_t*) data) = w > r ? w - r : 0;
|
||||
*((pa_usec_t*) data) = sink_get_latency(u);
|
||||
return 0;
|
||||
}
|
||||
|
||||
case SINK_MESSAGE_PASS_SOCKET: {
|
||||
case SINK_MESSAGE_TCP_PASS_SOCKET: {
|
||||
struct pollfd *pollfd;
|
||||
|
||||
pa_assert(!u->rtpoll_item);
|
||||
|
||||
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
|
||||
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
|
||||
pollfd->fd = u->fd;
|
||||
pollfd->fd = u->tcp_fd;
|
||||
pollfd->events = POLLOUT;
|
||||
/*pollfd->events = */pollfd->revents = 0;
|
||||
|
||||
if (u->sink->thread_info.state == PA_SINK_SUSPENDED) {
|
||||
/* Our stream has been suspended so we just flush it... */
|
||||
pa_raop_flush(u->raop);
|
||||
pa_raop_client_flush(u->raop);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
case SINK_MESSAGE_RIP_SOCKET: {
|
||||
if (u->fd >= 0) {
|
||||
pa_close(u->fd);
|
||||
u->fd = -1;
|
||||
case SINK_MESSAGE_TCP_RIP_SOCKET: {
|
||||
if (u->tcp_fd >= 0) {
|
||||
pa_close(u->tcp_fd);
|
||||
u->tcp_fd = -1;
|
||||
} else
|
||||
/* FIXME */
|
||||
pa_log("We should not get to this state. Cannot rip socket if not connected.");
|
||||
|
|
@ -260,10 +283,140 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
|
|||
return pa_sink_process_msg(o, code, data, offset, chunk);
|
||||
}
|
||||
|
||||
static void udp_start_wakeup_clock(struct userdata *u) {
|
||||
pa_usec_t now = pa_rtclock_now();
|
||||
|
||||
u->udp_playback_start = now;
|
||||
u->udp_sent_packets = 0;
|
||||
pa_rtpoll_set_timer_absolute(u->rtpoll, now);
|
||||
}
|
||||
|
||||
static pa_usec_t udp_next_wakeup_clock(struct userdata *u) {
|
||||
pa_usec_t intvl = pa_bytes_to_usec(u->block_size * u->udp_sent_packets,
|
||||
&u->sink->sample_spec);
|
||||
/* FIXME: how long until (u->block_size * u->udp_sent_packets) wraps?? */
|
||||
|
||||
return u->udp_playback_start + intvl;
|
||||
}
|
||||
|
||||
static int udp_sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
|
||||
struct userdata *u = PA_SINK(o)->userdata;
|
||||
|
||||
switch (code) {
|
||||
case PA_SINK_MESSAGE_SET_STATE:
|
||||
switch ((pa_sink_state_t) PA_PTR_TO_UINT(data)) {
|
||||
case PA_SINK_SUSPENDED:
|
||||
pa_assert(PA_SINK_IS_OPENED(u->sink->thread_info.state));
|
||||
pa_log_debug("RAOP: SUSPENDED");
|
||||
pa_smoother_pause(u->smoother, pa_rtclock_now());
|
||||
|
||||
if (pa_raop_client_udp_can_stream(u->raop)) {
|
||||
/* Issue a TEARDOWN if we are still connected. */
|
||||
pa_raop_client_teardown(u->raop);
|
||||
}
|
||||
|
||||
break;
|
||||
|
||||
case PA_SINK_IDLE:
|
||||
pa_log_debug("RAOP: IDLE");
|
||||
/* Issue a flush if we're comming from running state. */
|
||||
if (u->sink->thread_info.state == PA_SINK_RUNNING) {
|
||||
pa_rtpoll_set_timer_disabled(u->rtpoll);
|
||||
pa_raop_client_flush(u->raop);
|
||||
}
|
||||
|
||||
break;
|
||||
|
||||
case PA_SINK_RUNNING:
|
||||
pa_log_debug("RAOP: RUNNING");
|
||||
|
||||
pa_smoother_resume(u->smoother, pa_rtclock_now(), true);
|
||||
|
||||
if (!pa_raop_client_udp_can_stream(u->raop)) {
|
||||
/* Connecting will trigger a RECORD */
|
||||
pa_raop_client_connect(u->raop);
|
||||
}
|
||||
udp_start_wakeup_clock(u);
|
||||
|
||||
break;
|
||||
|
||||
case PA_SINK_UNLINKED:
|
||||
case PA_SINK_INIT:
|
||||
case PA_SINK_INVALID_STATE:
|
||||
;
|
||||
}
|
||||
|
||||
break;
|
||||
|
||||
case PA_SINK_MESSAGE_GET_LATENCY: {
|
||||
pa_usec_t r = 0;
|
||||
|
||||
if (pa_raop_client_udp_can_stream(u->raop))
|
||||
r = sink_get_latency(u);
|
||||
|
||||
*((pa_usec_t*) data) = r;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
case SINK_MESSAGE_UDP_SETUP: {
|
||||
struct pollfd *pollfd;
|
||||
|
||||
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 2);
|
||||
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
|
||||
|
||||
pollfd->fd = u->udp_control_fd;
|
||||
pollfd->events = POLLIN | POLLPRI;
|
||||
pollfd->revents = 0;
|
||||
pollfd++;
|
||||
pollfd->fd = u->udp_timing_fd;
|
||||
pollfd->events = POLLIN | POLLPRI;
|
||||
pollfd->revents = 0;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
case SINK_MESSAGE_UDP_RECORD: {
|
||||
udp_start_wakeup_clock(u);
|
||||
|
||||
if (u->sink->thread_info.state == PA_SINK_SUSPENDED) {
|
||||
/* Our stream has been suspended so we just flush it... */
|
||||
pa_rtpoll_set_timer_disabled(u->rtpoll);
|
||||
pa_raop_client_flush(u->raop);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
case SINK_MESSAGE_UDP_DISCONNECTED: {
|
||||
if (u->sink->thread_info.state == PA_SINK_SUSPENDED) {
|
||||
pa_rtpoll_set_timer_disabled(u->rtpoll);
|
||||
if (u->rtpoll_item)
|
||||
pa_rtpoll_item_free(u->rtpoll_item);
|
||||
u->rtpoll_item = NULL;
|
||||
} else {
|
||||
/* Question: is this valid here: or should we do some sort of:
|
||||
* return pa_sink_process_msg(PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL); ?? */
|
||||
pa_module_unload_request(u->module, true);
|
||||
}
|
||||
|
||||
pa_close(u->udp_control_fd);
|
||||
pa_close(u->udp_timing_fd);
|
||||
|
||||
u->udp_control_fd = -1;
|
||||
u->udp_timing_fd = -1;
|
||||
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
return pa_sink_process_msg(o, code, data, offset, chunk);
|
||||
}
|
||||
|
||||
static void sink_set_volume_cb(pa_sink *s) {
|
||||
struct userdata *u = s->userdata;
|
||||
pa_cvolume hw;
|
||||
pa_volume_t v;
|
||||
pa_volume_t v, v_orig;
|
||||
char t[PA_CVOLUME_SNPRINT_VERBOSE_MAX];
|
||||
|
||||
pa_assert(u);
|
||||
|
|
@ -277,11 +430,16 @@ static void sink_set_volume_cb(pa_sink *s) {
|
|||
* any variation in channel volumes in software. */
|
||||
v = pa_cvolume_max(&s->real_volume);
|
||||
|
||||
v_orig = v;
|
||||
v = pa_raop_client_adjust_volume(u->raop, v_orig);
|
||||
|
||||
pa_log_debug("Volume adjusted: orig=%u adjusted=%u", v_orig, v);
|
||||
|
||||
/* Create a pa_cvolume version of our single value. */
|
||||
pa_cvolume_set(&hw, s->sample_spec.channels, v);
|
||||
|
||||
/* Perform any software manipulation of the volume needed. */
|
||||
pa_sw_cvolume_divide(&s->soft_volume, &s->real_volume, &hw);
|
||||
/* Set the real volume based on given original volume. */
|
||||
pa_cvolume_set(&s->real_volume, s->sample_spec.channels, v_orig);
|
||||
|
||||
pa_log_debug("Requested volume: %s", pa_cvolume_snprint_verbose(t, sizeof(t), &s->real_volume, &s->channel_map, false));
|
||||
pa_log_debug("Got hardware volume: %s", pa_cvolume_snprint_verbose(t, sizeof(t), &hw, &s->channel_map, false));
|
||||
|
|
@ -305,8 +463,50 @@ static void sink_set_mute_cb(pa_sink *s) {
|
|||
}
|
||||
}
|
||||
|
||||
static void thread_func(void *userdata) {
|
||||
static void udp_setup_cb(int control_fd, int timing_fd, void *userdata) {
|
||||
struct userdata *u = userdata;
|
||||
|
||||
pa_assert(control_fd);
|
||||
pa_assert(timing_fd);
|
||||
pa_assert(u);
|
||||
|
||||
u->udp_control_fd = control_fd;
|
||||
u->udp_timing_fd = timing_fd;
|
||||
|
||||
pa_log_debug("Connection authenticated, syncing with server...");
|
||||
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UDP_SETUP, NULL, 0, NULL, NULL);
|
||||
}
|
||||
|
||||
static void udp_record_cb(void *userdata) {
|
||||
struct userdata *u = userdata;
|
||||
|
||||
pa_assert(u);
|
||||
|
||||
/* Set the initial volume. */
|
||||
sink_set_volume_cb(u->sink);
|
||||
|
||||
pa_log_debug("Synchronization done, pushing job to IO thread...");
|
||||
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UDP_RECORD, NULL, 0, NULL, NULL);
|
||||
}
|
||||
|
||||
static void udp_disconnected_cb(void *userdata) {
|
||||
struct userdata *u = userdata;
|
||||
|
||||
pa_assert(u);
|
||||
|
||||
/* This callback function is called from both STATE_TEARDOWN and
|
||||
STATE_DISCONNECTED in raop_client.c */
|
||||
|
||||
pa_assert(u);
|
||||
|
||||
pa_log_debug("Connection closed, informing IO thread...");
|
||||
|
||||
pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UDP_DISCONNECTED, NULL, 0, NULL, NULL);
|
||||
}
|
||||
|
||||
static void tcp_thread_func(struct userdata *u) {
|
||||
int write_type = 0;
|
||||
pa_memchunk silence;
|
||||
uint32_t silence_overhead = 0;
|
||||
|
|
@ -314,7 +514,7 @@ static void thread_func(void *userdata) {
|
|||
|
||||
pa_assert(u);
|
||||
|
||||
pa_log_debug("Thread starting up");
|
||||
pa_log_debug("TCP thread starting up");
|
||||
|
||||
pa_thread_mq_install(&u->thread_mq);
|
||||
|
||||
|
|
@ -394,7 +594,7 @@ static void thread_func(void *userdata) {
|
|||
pa_assert(u->encoded_memchunk.length > 0);
|
||||
|
||||
p = pa_memblock_acquire(u->encoded_memchunk.memblock);
|
||||
l = pa_write(u->fd, (uint8_t*) p + u->encoded_memchunk.index, u->encoded_memchunk.length, &write_type);
|
||||
l = pa_write(u->tcp_fd, (uint8_t*) p + u->encoded_memchunk.index, u->encoded_memchunk.length, &write_type);
|
||||
pa_memblock_release(u->encoded_memchunk.memblock);
|
||||
|
||||
pa_assert(l != 0);
|
||||
|
|
@ -443,7 +643,7 @@ static void thread_func(void *userdata) {
|
|||
#ifdef SIOCOUTQ
|
||||
{
|
||||
int l;
|
||||
if (ioctl(u->fd, SIOCOUTQ, &l) >= 0 && l > 0)
|
||||
if (ioctl(u->tcp_fd, SIOCOUTQ, &l) >= 0 && l > 0)
|
||||
n -= (l / u->encoding_ratio);
|
||||
}
|
||||
#endif
|
||||
|
|
@ -497,15 +697,139 @@ fail:
|
|||
finish:
|
||||
if (silence.memblock)
|
||||
pa_memblock_unref(silence.memblock);
|
||||
pa_log_debug("Thread shutting down");
|
||||
pa_log_debug("TCP thread shutting down");
|
||||
}
|
||||
|
||||
static void udp_thread_func(struct userdata *u) {
|
||||
pa_assert(u);
|
||||
|
||||
pa_log_debug("UDP thread starting up");
|
||||
|
||||
pa_thread_mq_install(&u->thread_mq);
|
||||
pa_smoother_set_time_offset(u->smoother, pa_rtclock_now());
|
||||
|
||||
for (;;) {
|
||||
pa_usec_t estimated;
|
||||
int32_t overhead = 0;
|
||||
ssize_t written = 0;
|
||||
size_t length = 0;
|
||||
int rv = 0;
|
||||
|
||||
if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
|
||||
if (u->sink->thread_info.rewind_requested)
|
||||
pa_sink_process_rewind(u->sink, 0);
|
||||
}
|
||||
|
||||
/* Polling (audio data + control socket + timing socket). */
|
||||
if ((rv = pa_rtpoll_run(u->rtpoll)) < 0)
|
||||
goto fail;
|
||||
else if (rv == 0)
|
||||
goto finish;
|
||||
|
||||
if (!pa_rtpoll_timer_elapsed(u->rtpoll)) {
|
||||
struct pollfd *pollfd;
|
||||
uint8_t packet[32];
|
||||
ssize_t read;
|
||||
|
||||
if (u->rtpoll_item) {
|
||||
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
|
||||
|
||||
/* Event on the control socket ?? */
|
||||
if (pollfd->revents & POLLIN) {
|
||||
pollfd->revents = 0;
|
||||
pa_log_debug("Received control packet.");
|
||||
read = pa_read(pollfd->fd, packet, sizeof(packet), NULL);
|
||||
pa_raop_client_udp_handle_control_packet(u->raop, packet, read);
|
||||
}
|
||||
|
||||
pollfd++;
|
||||
|
||||
/* Event on the timing port ?? */
|
||||
if (pollfd->revents & POLLIN) {
|
||||
pollfd->revents = 0;
|
||||
pa_log_debug("Received timing packet.");
|
||||
read = pa_read(pollfd->fd, packet, sizeof(packet), NULL);
|
||||
pa_raop_client_udp_handle_timing_packet(u->raop, packet, read);
|
||||
}
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!pa_raop_client_udp_can_stream(u->raop))
|
||||
continue;
|
||||
if (u->sink->thread_info.state != PA_SINK_RUNNING)
|
||||
continue;
|
||||
|
||||
if (u->encoded_memchunk.length <= 0) {
|
||||
if (u->encoded_memchunk.memblock != NULL)
|
||||
pa_memblock_unref(u->encoded_memchunk.memblock);
|
||||
|
||||
if (u->raw_memchunk.length <= 0) {
|
||||
if (u->raw_memchunk.memblock)
|
||||
pa_memblock_unref(u->raw_memchunk.memblock);
|
||||
pa_memchunk_reset(&u->raw_memchunk);
|
||||
|
||||
/* Grab unencoded audio data from PulseAudio. */
|
||||
pa_sink_render_full(u->sink, u->block_size, &u->raw_memchunk);
|
||||
}
|
||||
|
||||
pa_assert(u->raw_memchunk.length > 0);
|
||||
|
||||
length = u->raw_memchunk.length;
|
||||
pa_raop_client_encode_sample(u->raop, &u->raw_memchunk, &u->encoded_memchunk);
|
||||
u->encoding_ratio = (double) u->encoded_memchunk.length / (double) (length - u->raw_memchunk.length);
|
||||
overhead = u->encoded_memchunk.length - (length - u->raw_memchunk.length);
|
||||
}
|
||||
|
||||
pa_assert(u->encoded_memchunk.length > 0);
|
||||
|
||||
written = pa_raop_client_udp_send_audio_packet(u->raop,&u->encoded_memchunk);
|
||||
if (written < 0) {
|
||||
pa_log("Failed to send UDP packet: %s", pa_cstrerror(errno));
|
||||
goto fail;
|
||||
}
|
||||
|
||||
u->udp_sent_packets++;
|
||||
/* Sleep until next packet transmission */
|
||||
pa_rtpoll_set_timer_absolute(u->rtpoll, udp_next_wakeup_clock(u));
|
||||
|
||||
u->offset += written;
|
||||
u->encoding_overhead += overhead;
|
||||
|
||||
estimated = pa_bytes_to_usec(u->offset - u->encoding_overhead, &u->sink->sample_spec);
|
||||
pa_smoother_put(u->smoother, pa_rtclock_now(), estimated);
|
||||
}
|
||||
|
||||
fail:
|
||||
/* If this was no regular exit, continue processing messages until PA_MESSAGE_SHUTDOWN. */
|
||||
pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
|
||||
pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
|
||||
|
||||
finish:
|
||||
pa_log_debug("UDP thread shutting down");
|
||||
}
|
||||
|
||||
static void thread_func(void *userdata) {
|
||||
struct userdata *u = userdata;
|
||||
|
||||
if (u->protocol == RAOP_TCP)
|
||||
tcp_thread_func(u);
|
||||
else if (u->protocol == RAOP_UDP)
|
||||
udp_thread_func(u);
|
||||
else
|
||||
pa_assert(false);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
int pa__init(pa_module *m) {
|
||||
struct userdata *u = NULL;
|
||||
pa_sample_spec ss;
|
||||
pa_modargs *ma = NULL;
|
||||
const char *server;
|
||||
const char *server, *protocol, *encryption;
|
||||
pa_sink_new_data data;
|
||||
char *t = NULL;
|
||||
|
||||
pa_assert(m);
|
||||
|
||||
|
|
@ -532,7 +856,7 @@ int pa__init(pa_module *m) {
|
|||
u->core = m->core;
|
||||
u->module = m;
|
||||
m->userdata = u;
|
||||
u->fd = -1;
|
||||
u->tcp_fd = -1;
|
||||
u->smoother = pa_smoother_new(
|
||||
PA_USEC_PER_SEC,
|
||||
PA_USEC_PER_SEC*2,
|
||||
|
|
@ -574,15 +898,32 @@ int pa__init(pa_module *m) {
|
|||
goto fail;
|
||||
}
|
||||
|
||||
/* This may be overwriten if sink_name is specified in module arguments. */
|
||||
t = pa_sprintf_malloc("raop_client.%s", server);
|
||||
|
||||
protocol = pa_modargs_get_value(ma, "protocol", NULL);
|
||||
if (protocol == NULL || pa_streq(protocol, "TCP")) {
|
||||
/* Assume TCP by default */
|
||||
u->protocol = RAOP_TCP;
|
||||
}
|
||||
else if (pa_streq(protocol, "UDP")) {
|
||||
u->protocol = RAOP_UDP;
|
||||
} else {
|
||||
pa_log("Unsupported protocol argument given: %s", protocol);
|
||||
goto fail;
|
||||
}
|
||||
|
||||
pa_sink_new_data_init(&data);
|
||||
data.driver = __FILE__;
|
||||
data.module = m;
|
||||
pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
|
||||
pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", t));
|
||||
pa_sink_new_data_set_sample_spec(&data, &ss);
|
||||
pa_proplist_sets(data.proplist, PA_PROP_DEVICE_STRING, server);
|
||||
pa_proplist_sets(data.proplist, PA_PROP_DEVICE_INTENDED_ROLES, "music");
|
||||
pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "RAOP sink '%s'", server);
|
||||
|
||||
/* RAOP discover module will eventually overwrite sink_name and others
|
||||
(PA_UPDATE_REPLACE). */
|
||||
if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
|
||||
pa_log("Invalid properties");
|
||||
pa_sink_new_data_done(&data);
|
||||
|
|
@ -590,6 +931,7 @@ int pa__init(pa_module *m) {
|
|||
}
|
||||
|
||||
u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY|PA_SINK_NETWORK);
|
||||
pa_xfree(t); t = NULL;
|
||||
pa_sink_new_data_done(&data);
|
||||
|
||||
if (!u->sink) {
|
||||
|
|
@ -597,7 +939,10 @@ int pa__init(pa_module *m) {
|
|||
goto fail;
|
||||
}
|
||||
|
||||
u->sink->parent.process_msg = sink_process_msg;
|
||||
if (u->protocol == RAOP_TCP)
|
||||
u->sink->parent.process_msg = tcp_sink_process_msg;
|
||||
else
|
||||
u->sink->parent.process_msg = udp_sink_process_msg;
|
||||
u->sink->userdata = u;
|
||||
pa_sink_set_set_volume_callback(u->sink, sink_set_volume_cb);
|
||||
pa_sink_set_set_mute_callback(u->sink, sink_set_mute_cb);
|
||||
|
|
@ -606,13 +951,27 @@ int pa__init(pa_module *m) {
|
|||
pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
|
||||
pa_sink_set_rtpoll(u->sink, u->rtpoll);
|
||||
|
||||
if (!(u->raop = pa_raop_client_new(u->core, server))) {
|
||||
if (!(u->raop = pa_raop_client_new(u->core, server, u->protocol))) {
|
||||
pa_log("Failed to connect to server.");
|
||||
goto fail;
|
||||
}
|
||||
|
||||
pa_raop_client_set_callback(u->raop, on_connection, u);
|
||||
pa_raop_client_set_closed_callback(u->raop, on_close, u);
|
||||
encryption = pa_modargs_get_value(ma, "encryption", NULL);
|
||||
pa_raop_client_set_encryption(u->raop, !pa_safe_streq(encryption, "none"));
|
||||
|
||||
pa_raop_client_tcp_set_callback(u->raop, tcp_on_connection, u);
|
||||
pa_raop_client_tcp_set_closed_callback(u->raop, tcp_on_close, u);
|
||||
|
||||
if (u->protocol == RAOP_UDP) {
|
||||
/* The number of frames per blocks is not negotiable... */
|
||||
pa_raop_client_udp_get_blocks_size(u->raop, &u->block_size);
|
||||
u->block_size *= pa_frame_size(&ss);
|
||||
pa_sink_set_max_request(u->sink, u->block_size);
|
||||
|
||||
pa_raop_client_udp_set_setup_callback(u->raop, udp_setup_cb, u);
|
||||
pa_raop_client_udp_set_record_callback(u->raop, udp_record_cb, u);
|
||||
pa_raop_client_udp_set_disconnected_callback(u->raop, udp_disconnected_cb, u);
|
||||
}
|
||||
|
||||
if (!(u->thread = pa_thread_new("raop-sink", thread_func, u))) {
|
||||
pa_log("Failed to create thread.");
|
||||
|
|
@ -626,6 +985,8 @@ int pa__init(pa_module *m) {
|
|||
return 0;
|
||||
|
||||
fail:
|
||||
pa_xfree(t);
|
||||
|
||||
if (ma)
|
||||
pa_modargs_free(ma);
|
||||
|
||||
|
|
@ -684,8 +1045,8 @@ void pa__done(pa_module *m) {
|
|||
if (u->smoother)
|
||||
pa_smoother_free(u->smoother);
|
||||
|
||||
if (u->fd >= 0)
|
||||
pa_close(u->fd);
|
||||
if (u->tcp_fd >= 0)
|
||||
pa_close(u->tcp_fd);
|
||||
|
||||
pa_xfree(u);
|
||||
}
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -20,23 +20,52 @@
|
|||
along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
|
||||
***/
|
||||
|
||||
#include <pulse/volume.h>
|
||||
|
||||
#include <pulsecore/core.h>
|
||||
#include <pulsecore/memchunk.h>
|
||||
|
||||
typedef enum pa_raop_protocol {
|
||||
RAOP_TCP,
|
||||
RAOP_UDP,
|
||||
} pa_raop_protocol_t;
|
||||
|
||||
typedef struct pa_raop_client pa_raop_client;
|
||||
|
||||
pa_raop_client* pa_raop_client_new(pa_core *core, const char *host);
|
||||
pa_raop_client* pa_raop_client_new(pa_core *core, const char *host, pa_raop_protocol_t protocol);
|
||||
void pa_raop_client_free(pa_raop_client *c);
|
||||
|
||||
int pa_raop_connect(pa_raop_client *c);
|
||||
int pa_raop_flush(pa_raop_client *c);
|
||||
int pa_raop_client_connect(pa_raop_client *c);
|
||||
int pa_raop_client_flush(pa_raop_client *c);
|
||||
int pa_raop_client_teardown(pa_raop_client *c);
|
||||
|
||||
int pa_raop_client_udp_can_stream(pa_raop_client *c);
|
||||
|
||||
void pa_raop_client_set_encryption(pa_raop_client *c, int encryption);
|
||||
pa_volume_t pa_raop_client_adjust_volume(pa_raop_client *c, pa_volume_t volume);
|
||||
int pa_raop_client_set_volume(pa_raop_client *c, pa_volume_t volume);
|
||||
int pa_raop_client_encode_sample(pa_raop_client *c, pa_memchunk *raw, pa_memchunk *encoded);
|
||||
|
||||
int pa_raop_client_udp_handle_timing_packet(pa_raop_client *c, const uint8_t packet
|
||||
[], ssize_t size);
|
||||
int pa_raop_client_udp_handle_control_packet(pa_raop_client *c, const uint8_t packet[], ssize_t size);
|
||||
int pa_raop_client_udp_get_blocks_size(pa_raop_client *c, size_t *size);
|
||||
ssize_t pa_raop_client_udp_send_audio_packet(pa_raop_client *c, pa_memchunk *block);
|
||||
|
||||
typedef void (*pa_raop_client_cb_t)(int fd, void *userdata);
|
||||
void pa_raop_client_set_callback(pa_raop_client *c, pa_raop_client_cb_t callback, void *userdata);
|
||||
void pa_raop_client_tcp_set_callback(pa_raop_client *c, pa_raop_client_cb_t callback, void *userdata);
|
||||
|
||||
typedef void (*pa_raop_client_closed_cb_t)(void *userdata);
|
||||
void pa_raop_client_set_closed_callback(pa_raop_client *c, pa_raop_client_closed_cb_t callback, void *userdata);
|
||||
void pa_raop_client_tcp_set_closed_callback(pa_raop_client *c, pa_raop_client_closed_cb_t callback, void *userdata);
|
||||
|
||||
|
||||
typedef void (*pa_raop_client_setup_cb_t)(int control_fd, int timing_fd, void *userdata);
|
||||
void pa_raop_client_udp_set_setup_callback(pa_raop_client *c, pa_raop_client_setup_cb_t callback, void *userdata);
|
||||
|
||||
typedef void (*pa_raop_client_record_cb_t)(void *userdata);
|
||||
void pa_raop_client_udp_set_record_callback(pa_raop_client *c, pa_raop_client_record_cb_t callback, void *userdata);
|
||||
|
||||
typedef void (*pa_raop_client_disconnected_cb_t)(void *userdata);
|
||||
void pa_raop_client_udp_set_disconnected_callback(pa_raop_client *c, pa_raop_client_disconnected_cb_t callback, void *userdata);
|
||||
|
||||
#endif
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue