aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/decoder/mpcdec_decoder_plugin.c4
-rw-r--r--src/input/curl_input_plugin.c30
-rw-r--r--src/output/httpd_client.c27
-rw-r--r--src/output/pulse_output_plugin.c67
4 files changed, 122 insertions, 6 deletions
diff --git a/src/decoder/mpcdec_decoder_plugin.c b/src/decoder/mpcdec_decoder_plugin.c
index d28c22305..f31dcdb99 100644
--- a/src/decoder/mpcdec_decoder_plugin.c
+++ b/src/decoder/mpcdec_decoder_plugin.c
@@ -153,7 +153,6 @@ mpcdec_decode(struct decoder *mpd_decoder, struct input_stream *is)
mpc_uint32_t ret;
int32_t chunk[G_N_ELEMENTS(sample_buffer)];
long bit_rate = 0;
- mpc_uint32_t vbr_update_acc;
mpc_uint32_t vbr_update_bits;
enum decoder_command cmd = DECODE_COMMAND_NONE;
@@ -243,10 +242,11 @@ mpcdec_decode(struct decoder *mpd_decoder, struct input_stream *is)
decoder_seek_error(mpd_decoder);
}
- vbr_update_acc = 0;
vbr_update_bits = 0;
#ifdef MPC_IS_OLD_API
+ mpc_uint32_t vbr_update_acc = 0;
+
ret = mpc_decoder_decode(&decoder, sample_buffer,
&vbr_update_acc, &vbr_update_bits);
if (ret == 0 || ret == (mpc_uint32_t)-1)
diff --git a/src/input/curl_input_plugin.c b/src/input/curl_input_plugin.c
index 09010c220..20e8dea07 100644
--- a/src/input/curl_input_plugin.c
+++ b/src/input/curl_input_plugin.c
@@ -43,6 +43,13 @@
#define G_LOG_DOMAIN "input_curl"
/**
+ * Do not buffer more than this number of bytes. It should be a
+ * reasonable limit that doesn't make low-end machines suffer too
+ * much, but doesn't cause stuttering on high-latency lines.
+ */
+static const size_t CURL_MAX_BUFFERED = 512 * 1024;
+
+/**
* Buffers created by input_curl_writefunction().
*/
struct buffer {
@@ -144,6 +151,25 @@ input_curl_finish(void)
curl_global_cleanup();
}
+/**
+ * Determine the total sizes of all buffers, including portions that
+ * have already been consumed.
+ */
+G_GNUC_PURE
+static size_t
+curl_total_buffer_size(const struct input_curl *c)
+{
+ size_t total = 0;
+
+ for (GList *i = g_queue_peek_head_link(c->buffers);
+ i != NULL; i = g_list_next(i)) {
+ struct buffer *buffer = i->data;
+ total += buffer->size;
+ }
+
+ return total;
+}
+
static void
buffer_free_callback(gpointer data, G_GNUC_UNUSED gpointer user_data)
{
@@ -473,6 +499,10 @@ static int
input_curl_buffer(struct input_stream *is, GError **error_r)
{
struct input_curl *c = (struct input_curl *)is;
+
+ if (curl_total_buffer_size(c) >= CURL_MAX_BUFFERED)
+ return 0;
+
CURLMcode mcode;
int running_handles;
bool ret;
diff --git a/src/output/httpd_client.c b/src/output/httpd_client.c
index 32f355a43..e2c49c6c8 100644
--- a/src/output/httpd_client.c
+++ b/src/output/httpd_client.c
@@ -143,6 +143,8 @@ httpd_client_unref_page(gpointer data, G_GNUC_UNUSED gpointer user_data)
void
httpd_client_free(struct httpd_client *client)
{
+ assert(client != NULL);
+
if (client->state == RESPONSE) {
if (client->write_source_id != 0)
g_source_remove(client->write_source_id);
@@ -169,6 +171,8 @@ httpd_client_free(struct httpd_client *client)
static void
httpd_client_close(struct httpd_client *client)
{
+ assert(client != NULL);
+
httpd_output_remove_client(client->httpd, client);
httpd_client_free(client);
}
@@ -179,6 +183,9 @@ httpd_client_close(struct httpd_client *client)
static void
httpd_client_begin_response(struct httpd_client *client)
{
+ assert(client != NULL);
+ assert(client->state != RESPONSE);
+
client->state = RESPONSE;
client->write_source_id = 0;
client->pages = g_queue_new();
@@ -239,6 +246,9 @@ httpd_client_handle_line(struct httpd_client *client, const char *line)
static char *
httpd_client_read_line(struct httpd_client *client)
{
+ assert(client != NULL);
+ assert(client->state != RESPONSE);
+
const char *p, *newline;
size_t length;
char *line;
@@ -271,6 +281,7 @@ httpd_client_send_response(struct httpd_client *client)
GIOStatus status;
gsize bytes_written;
+ assert(client != NULL);
assert(client->state == RESPONSE);
if (!client->metadata_requested) {
@@ -334,14 +345,19 @@ httpd_client_send_response(struct httpd_client *client)
static bool
httpd_client_received(struct httpd_client *client)
{
+ assert(client != NULL);
+ assert(client->state != RESPONSE);
+
char *line;
bool success;
while ((line = httpd_client_read_line(client)) != NULL) {
success = httpd_client_handle_line(client, line);
g_free(line);
- if (!success)
+ if (!success) {
+ assert(client->state != RESPONSE);
return false;
+ }
if (client->state == RESPONSE) {
if (!fifo_buffer_is_empty(client->input)) {
@@ -370,7 +386,14 @@ httpd_client_read(struct httpd_client *client)
if (client->state == RESPONSE) {
/* the client has already sent the request, and he
must not send more */
- g_warning("unexpected input from client");
+ char buffer[1];
+
+ status = g_io_channel_read_chars(client->channel, buffer,
+ sizeof(buffer), &bytes_read,
+ NULL);
+ if (status == G_IO_STATUS_NORMAL)
+ g_warning("unexpected input from client");
+
return false;
}
diff --git a/src/output/pulse_output_plugin.c b/src/output/pulse_output_plugin.c
index 70cfb9784..9fb0b3ade 100644
--- a/src/output/pulse_output_plugin.c
+++ b/src/output/pulse_output_plugin.c
@@ -207,6 +207,9 @@ pulse_output_subscribe_cb(pa_context *context,
static bool
pulse_output_connect(struct pulse_output *po, GError **error_r)
{
+ assert(po != NULL);
+ assert(po->context != NULL);
+
int error;
error = pa_context_connect(po->context, po->server,
@@ -229,6 +232,9 @@ pulse_output_connect(struct pulse_output *po, GError **error_r)
static bool
pulse_output_setup_context(struct pulse_output *po, GError **error_r)
{
+ assert(po != NULL);
+ assert(po->mainloop != NULL);
+
po->context = pa_context_new(pa_threaded_mainloop_get_api(po->mainloop),
MPD_PULSE_NAME);
if (po->context == NULL) {
@@ -257,6 +263,9 @@ pulse_output_setup_context(struct pulse_output *po, GError **error_r)
static void
pulse_output_delete_context(struct pulse_output *po)
{
+ assert(po != NULL);
+ assert(po->context != NULL);
+
pa_context_disconnect(po->context);
pa_context_unref(po->context);
po->context = NULL;
@@ -347,6 +356,8 @@ pulse_output_disable(void *data)
{
struct pulse_output *po = data;
+ assert(po->mainloop != NULL);
+
pa_threaded_mainloop_stop(po->mainloop);
if (po->context != NULL)
pulse_output_delete_context(po);
@@ -363,6 +374,8 @@ pulse_output_disable(void *data)
static bool
pulse_output_wait_connection(struct pulse_output *po, GError **error_r)
{
+ assert(po->mainloop != NULL);
+
pa_context_state_t state;
pa_threaded_mainloop_lock(po->mainloop);
@@ -399,11 +412,32 @@ pulse_output_wait_connection(struct pulse_output *po, GError **error_r)
}
}
+#if PA_CHECK_VERSION(0,9,8)
+
+static void
+pulse_output_stream_suspended_cb(G_GNUC_UNUSED pa_stream *stream, void *userdata)
+{
+ struct pulse_output *po = userdata;
+
+ assert(stream == po->stream || po->stream == NULL);
+ assert(po->mainloop != NULL);
+
+ /* wake up the main loop to break out of the loop in
+ pulse_output_play() */
+ pa_threaded_mainloop_signal(po->mainloop, 0);
+}
+
+#endif
+
static void
pulse_output_stream_state_cb(pa_stream *stream, void *userdata)
{
struct pulse_output *po = userdata;
+ assert(stream == po->stream || po->stream == NULL);
+ assert(po->mainloop != NULL);
+ assert(po->context != NULL);
+
switch (pa_stream_get_state(stream)) {
case PA_STREAM_READY:
if (po->mixer != NULL)
@@ -432,6 +466,8 @@ pulse_output_stream_write_cb(G_GNUC_UNUSED pa_stream *stream, size_t nbytes,
{
struct pulse_output *po = userdata;
+ assert(po->mainloop != NULL);
+
po->writable = nbytes;
pa_threaded_mainloop_signal(po->mainloop, 0);
}
@@ -444,6 +480,8 @@ pulse_output_open(void *data, struct audio_format *audio_format,
pa_sample_spec ss;
int error;
+ assert(po->mainloop != NULL);
+
if (po->context != NULL) {
switch (pa_context_get_state(po->context)) {
case PA_CONTEXT_UNCONNECTED:
@@ -487,6 +525,11 @@ pulse_output_open(void *data, struct audio_format *audio_format,
return false;
}
+#if PA_CHECK_VERSION(0,9,8)
+ pa_stream_set_suspended_callback(po->stream,
+ pulse_output_stream_suspended_cb, po);
+#endif
+
pa_stream_set_state_callback(po->stream,
pulse_output_stream_state_cb, po);
pa_stream_set_write_callback(po->stream,
@@ -522,6 +565,8 @@ pulse_output_close(void *data)
struct pulse_output *po = data;
pa_operation *o;
+ assert(po->mainloop != NULL);
+
pa_threaded_mainloop_lock(po->mainloop);
if (pa_stream_get_state(po->stream) == PA_STREAM_READY) {
@@ -556,6 +601,8 @@ pulse_output_check_stream(struct pulse_output *po)
{
pa_stream_state_t state = pa_stream_get_state(po->stream);
+ assert(po->mainloop != NULL);
+
switch (state) {
case PA_STREAM_READY:
case PA_STREAM_FAILED:
@@ -637,6 +684,8 @@ pulse_output_stream_pause(struct pulse_output *po, bool pause,
{
pa_operation *o;
+ assert(po->mainloop != NULL);
+ assert(po->context != NULL);
assert(po->stream != NULL);
o = pa_stream_cork(po->stream, pause,
@@ -667,6 +716,7 @@ pulse_output_play(void *data, const void *chunk, size_t size, GError **error_r)
struct pulse_output *po = data;
int error;
+ assert(po->mainloop != NULL);
assert(po->stream != NULL);
pa_threaded_mainloop_lock(po->mainloop);
@@ -683,19 +733,30 @@ pulse_output_play(void *data, const void *chunk, size_t size, GError **error_r)
/* unpause if previously paused */
if (pulse_output_stream_is_paused(po) &&
- !pulse_output_stream_pause(po, false, error_r))
+ !pulse_output_stream_pause(po, false, error_r)) {
+ pa_threaded_mainloop_unlock(po->mainloop);
return 0;
+ }
/* wait until the server allows us to write */
while (po->writable == 0) {
+#if PA_CHECK_VERSION(0,9,8)
+ if (pa_stream_is_suspended(po->stream)) {
+ pa_threaded_mainloop_unlock(po->mainloop);
+ g_set_error(error_r, pulse_output_quark(), 0,
+ "suspended");
+ return 0;
+ }
+#endif
+
pa_threaded_mainloop_wait(po->mainloop);
if (pa_stream_get_state(po->stream) != PA_STREAM_READY) {
pa_threaded_mainloop_unlock(po->mainloop);
g_set_error(error_r, pulse_output_quark(), 0,
"disconnected");
- return false;
+ return 0;
}
}
@@ -725,6 +786,7 @@ pulse_output_cancel(void *data)
struct pulse_output *po = data;
pa_operation *o;
+ assert(po->mainloop != NULL);
assert(po->stream != NULL);
pa_threaded_mainloop_lock(po->mainloop);
@@ -756,6 +818,7 @@ pulse_output_pause(void *data)
struct pulse_output *po = data;
GError *error = NULL;
+ assert(po->mainloop != NULL);
assert(po->stream != NULL);
pa_threaded_mainloop_lock(po->mainloop);
generated by cgit v1.2.3 (git 2.25.1) at 2024-12-19 12:06:40 +0000