aboutsummaryrefslogtreecommitdiffstats
path: root/src/input
diff options
context:
space:
mode:
Diffstat (limited to 'src/input')
-rw-r--r--src/input/archive_input_plugin.c7
-rw-r--r--src/input/cdio_paranoia_input_plugin.c7
-rw-r--r--src/input/curl_input_plugin.c108
-rw-r--r--src/input/despotify_input_plugin.c7
-rw-r--r--src/input/ffmpeg_input_plugin.c7
-rw-r--r--src/input/file_input_plugin.c7
-rw-r--r--src/input/mms_input_plugin.c7
-rw-r--r--src/input/rewind_input_plugin.c15
-rw-r--r--src/input/soup_input_plugin.c102
9 files changed, 101 insertions, 166 deletions
diff --git a/src/input/archive_input_plugin.c b/src/input/archive_input_plugin.c
index 8d78f4c89..4a038b9e2 100644
--- a/src/input/archive_input_plugin.c
+++ b/src/input/archive_input_plugin.c
@@ -34,7 +34,9 @@
* plugin and gzip fetches file from disk
*/
static struct input_stream *
-input_archive_open(const char *pathname, GError **error_r)
+input_archive_open(const char *pathname,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
const struct archive_plugin *arplug;
struct archive_file *file;
@@ -65,7 +67,8 @@ input_archive_open(const char *pathname, GError **error_r)
return NULL;
//setup fileops
- is = archive_file_open_stream(file, filename, error_r);
+ is = archive_file_open_stream(file, filename, mutex, cond,
+ error_r);
archive_file_close(file);
g_free(pname);
diff --git a/src/input/cdio_paranoia_input_plugin.c b/src/input/cdio_paranoia_input_plugin.c
index da2ed1096..1a1c8d2c5 100644
--- a/src/input/cdio_paranoia_input_plugin.c
+++ b/src/input/cdio_paranoia_input_plugin.c
@@ -149,7 +149,9 @@ cdio_detect_device(void)
}
static struct input_stream *
-input_cdio_open(const char *uri, GError **error_r)
+input_cdio_open(const char *uri,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
struct input_cdio_paranoia *i;
@@ -158,7 +160,8 @@ input_cdio_open(const char *uri, GError **error_r)
return NULL;
i = g_new(struct input_cdio_paranoia, 1);
- input_stream_init(&i->base, &input_plugin_cdio_paranoia, uri);
+ input_stream_init(&i->base, &input_plugin_cdio_paranoia, uri,
+ mutex, cond);
/* initialize everything (should be already) */
i->drv = NULL;
diff --git a/src/input/curl_input_plugin.c b/src/input/curl_input_plugin.c
index 4f3e9c6e4..88a5556d2 100644
--- a/src/input/curl_input_plugin.c
+++ b/src/input/curl_input_plugin.c
@@ -81,9 +81,6 @@ struct input_curl {
/** the curl handles */
CURL *easy;
- GMutex *mutex;
- GCond *cond;
-
/** the GMainLoop source used to poll all CURL file
descriptors */
GSource *source;
@@ -433,11 +430,11 @@ input_curl_abort_all_requests(GError *error)
input_curl_easy_free(c);
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
c->postponed_error = g_error_copy(error);
c->base.ready = true;
- g_cond_broadcast(c->cond);
- g_mutex_unlock(c->mutex);
+ g_cond_broadcast(c->base.cond);
+ g_mutex_unlock(c->base.mutex);
}
g_error_free(error);
@@ -457,7 +454,7 @@ input_curl_request_done(struct input_curl *c, CURLcode result, long status)
assert(c->easy == NULL);
assert(c->postponed_error == NULL);
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
if (result != CURLE_OK) {
c->postponed_error = g_error_new(curl_quark(), result,
@@ -470,8 +467,8 @@ input_curl_request_done(struct input_curl *c, CURLcode result, long status)
}
c->base.ready = true;
- g_cond_broadcast(c->cond);
- g_mutex_unlock(c->mutex);
+ g_cond_broadcast(c->base.cond);
+ g_mutex_unlock(c->base.mutex);
}
static void
@@ -763,9 +760,6 @@ input_curl_free(struct input_curl *c)
g_queue_free(c->buffers);
- g_mutex_free(c->mutex);
- g_cond_free(c->cond);
-
if (c->postponed_error != NULL)
g_error_free(c->postponed_error);
@@ -779,15 +773,12 @@ input_curl_check(struct input_stream *is, GError **error_r)
{
struct input_curl *c = (struct input_curl *)is;
- g_mutex_lock(c->mutex);
-
bool success = c->postponed_error == NULL;
if (!success) {
g_propagate_error(error_r, c->postponed_error);
c->postponed_error = NULL;
}
- g_mutex_unlock(c->mutex);
return success;
}
@@ -805,7 +796,7 @@ static bool
fill_buffer(struct input_curl *c, GError **error_r)
{
while (c->easy != NULL && g_queue_is_empty(c->buffers))
- g_cond_wait(c->cond, c->mutex);
+ g_cond_wait(c->base.cond, c->base.mutex);
if (c->postponed_error != NULL) {
g_propagate_error(error_r, c->postponed_error);
@@ -906,6 +897,15 @@ copy_icy_tag(struct input_curl *c)
c->tag = tag;
}
+static bool
+input_curl_available(struct input_stream *is)
+{
+ struct input_curl *c = (struct input_curl *)is;
+
+ return c->postponed_error != NULL || c->easy == NULL ||
+ !g_queue_is_empty(c->buffers);
+}
+
static size_t
input_curl_read(struct input_stream *is, void *ptr, size_t size,
GError **error_r)
@@ -915,16 +915,12 @@ input_curl_read(struct input_stream *is, void *ptr, size_t size,
size_t nbytes = 0;
char *dest = ptr;
- g_mutex_lock(c->mutex);
-
do {
/* fill the buffer */
success = fill_buffer(c, error_r);
- if (!success) {
- g_mutex_unlock(c->mutex);
+ if (!success)
return 0;
- }
/* send buffer contents */
@@ -944,14 +940,12 @@ input_curl_read(struct input_stream *is, void *ptr, size_t size,
#if LIBCURL_VERSION_NUM >= 0x071200
if (c->paused && curl_total_buffer_size(c) < CURL_RESUME_AT) {
- g_mutex_unlock(c->mutex);
+ g_mutex_unlock(c->base.mutex);
io_thread_call(input_curl_resume, c);
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
}
#endif
- g_mutex_unlock(c->mutex);
-
return nbytes;
}
@@ -968,33 +962,7 @@ input_curl_eof(G_GNUC_UNUSED struct input_stream *is)
{
struct input_curl *c = (struct input_curl *)is;
- g_mutex_lock(c->mutex);
- bool eof = c->easy == NULL && g_queue_is_empty(c->buffers);
- g_mutex_unlock(c->mutex);
-
- return eof;
-}
-
-static int
-input_curl_buffer(struct input_stream *is, GError **error_r)
-{
- struct input_curl *c = (struct input_curl *)is;
-
- g_mutex_lock(c->mutex);
-
- int result;
- if (c->postponed_error != NULL) {
- g_propagate_error(error_r, c->postponed_error);
- c->postponed_error = NULL;
- result = -1;
- } else if (g_queue_is_empty(c->buffers))
- result = 0;
- else
- result = 1;
-
- g_mutex_unlock(c->mutex);
-
- return result;
+ return c->easy == NULL && g_queue_is_empty(c->buffers);
}
/** called by curl when new data is available */
@@ -1092,12 +1060,12 @@ input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
if (size == 0)
return 0;
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
#if LIBCURL_VERSION_NUM >= 0x071200
if (curl_total_buffer_size(c) + size >= CURL_MAX_BUFFERED) {
c->paused = true;
- g_mutex_unlock(c->mutex);
+ g_mutex_unlock(c->base.mutex);
return CURL_WRITEFUNC_PAUSE;
}
#endif
@@ -1108,11 +1076,10 @@ input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
memcpy(buffer->data, ptr, size);
g_queue_push_tail(c->buffers, buffer);
-
c->base.ready = true;
- g_cond_broadcast(c->cond);
- g_mutex_unlock(c->mutex);
+ g_cond_broadcast(c->base.cond);
+ g_mutex_unlock(c->base.mutex);
return size;
}
@@ -1219,8 +1186,6 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
/* check if we can fast-forward the buffer */
- g_mutex_lock(c->mutex);
-
while (offset > is->offset && !g_queue_is_empty(c->buffers)) {
struct buffer *buffer;
size_t length;
@@ -1238,13 +1203,13 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
is->offset += length;
}
- g_mutex_unlock(c->mutex);
-
if (offset == is->offset)
return true;
/* close the old connection and open a new one */
+ g_mutex_unlock(c->base.mutex);
+
input_curl_easy_free_indirect(c);
input_curl_flush_buffers(c);
@@ -1272,36 +1237,35 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
if (!input_curl_easy_add_indirect(c, error_r))
return false;
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
while (!c->base.ready)
- g_cond_wait(c->cond, c->mutex);
+ g_cond_wait(c->base.cond, c->base.mutex);
if (c->postponed_error != NULL) {
g_propagate_error(error_r, c->postponed_error);
c->postponed_error = NULL;
- g_mutex_unlock(c->mutex);
return false;
}
- g_mutex_unlock(c->mutex);
-
return true;
}
static struct input_stream *
-input_curl_open(const char *url, GError **error_r)
+input_curl_open(const char *url, GMutex *mutex, GCond *cond,
+ GError **error_r)
{
+ assert(mutex != NULL);
+ assert(cond != NULL);
+
struct input_curl *c;
if (strncmp(url, "http://", 7) != 0)
return NULL;
c = g_new0(struct input_curl, 1);
- input_stream_init(&c->base, &input_plugin_curl, url);
-
- c->mutex = g_mutex_new();
- c->cond = g_cond_new();
+ input_stream_init(&c->base, &input_plugin_curl, url,
+ mutex, cond);
c->url = g_strdup(url);
c->buffers = g_queue_new();
@@ -1337,7 +1301,7 @@ const struct input_plugin input_plugin_curl = {
.close = input_curl_close,
.check = input_curl_check,
.tag = input_curl_tag,
- .buffer = input_curl_buffer,
+ .available = input_curl_available,
.read = input_curl_read,
.eof = input_curl_eof,
.seek = input_curl_seek,
diff --git a/src/input/despotify_input_plugin.c b/src/input/despotify_input_plugin.c
index ef78fb1e0..200a0afd6 100644
--- a/src/input/despotify_input_plugin.c
+++ b/src/input/despotify_input_plugin.c
@@ -97,7 +97,9 @@ static void callback(G_GNUC_UNUSED struct despotify_session* ds,
static struct input_stream *
-input_despotify_open(const char *url, G_GNUC_UNUSED GError **error_r)
+input_despotify_open(const char *url,
+ GMutex *mutex, GCond *cond,
+ G_GNUC_UNUSED GError **error_r)
{
struct input_despotify *ctx;
struct despotify_session *session;
@@ -131,7 +133,8 @@ input_despotify_open(const char *url, G_GNUC_UNUSED GError **error_r)
return NULL;
}
- input_stream_init(&ctx->base, &input_plugin_despotify, url);
+ input_stream_init(&ctx->base, &input_plugin_despotify, url,
+ mutex, cond);
ctx->session = session;
ctx->track = track;
ctx->tag = mpd_despotify_tag_from_track(track);
diff --git a/src/input/ffmpeg_input_plugin.c b/src/input/ffmpeg_input_plugin.c
index 6caa7ea04..1c64b52c1 100644
--- a/src/input/ffmpeg_input_plugin.c
+++ b/src/input/ffmpeg_input_plugin.c
@@ -74,7 +74,9 @@ input_ffmpeg_init(G_GNUC_UNUSED const struct config_param *param,
}
static struct input_stream *
-input_ffmpeg_open(const char *uri, GError **error_r)
+input_ffmpeg_open(const char *uri,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
struct input_ffmpeg *i;
@@ -87,7 +89,8 @@ input_ffmpeg_open(const char *uri, GError **error_r)
return NULL;
i = g_new(struct input_ffmpeg, 1);
- input_stream_init(&i->base, &input_plugin_ffmpeg, uri);
+ input_stream_init(&i->base, &input_plugin_ffmpeg, uri,
+ mutex, cond);
#if LIBAVFORMAT_VERSION_INT >= AV_VERSION_INT(53,1,0)
int ret = avio_open(&i->h, uri, AVIO_FLAG_READ);
diff --git a/src/input/file_input_plugin.c b/src/input/file_input_plugin.c
index 10b753d24..5ee3f200b 100644
--- a/src/input/file_input_plugin.c
+++ b/src/input/file_input_plugin.c
@@ -46,7 +46,9 @@ file_quark(void)
}
static struct input_stream *
-input_file_open(const char *filename, GError **error_r)
+input_file_open(const char *filename,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
int fd, ret;
struct stat st;
@@ -85,7 +87,8 @@ input_file_open(const char *filename, GError **error_r)
#endif
fis = g_new(struct file_input_stream, 1);
- input_stream_init(&fis->base, &input_plugin_file, filename);
+ input_stream_init(&fis->base, &input_plugin_file, filename,
+ mutex, cond);
fis->base.size = st.st_size;
fis->base.seekable = true;
diff --git a/src/input/mms_input_plugin.c b/src/input/mms_input_plugin.c
index eb3e5d26c..cff15125b 100644
--- a/src/input/mms_input_plugin.c
+++ b/src/input/mms_input_plugin.c
@@ -46,7 +46,9 @@ mms_quark(void)
}
static struct input_stream *
-input_mms_open(const char *url, GError **error_r)
+input_mms_open(const char *url,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
struct input_mms *m;
@@ -57,7 +59,8 @@ input_mms_open(const char *url, GError **error_r)
return NULL;
m = g_new(struct input_mms, 1);
- input_stream_init(&m->base, &input_plugin_mms, url);
+ input_stream_init(&m->base, &input_plugin_mms, url,
+ mutex, cond);
m->mms = mmsx_connect(NULL, NULL, url, 128 * 1024);
if (m->mms == NULL) {
diff --git a/src/input/rewind_input_plugin.c b/src/input/rewind_input_plugin.c
index 2a3eecf82..cf06fc57b 100644
--- a/src/input/rewind_input_plugin.c
+++ b/src/input/rewind_input_plugin.c
@@ -132,16 +132,12 @@ input_rewind_tag(struct input_stream *is)
return input_stream_tag(r->input);
}
-static int
-input_rewind_buffer(struct input_stream *is, GError **error_r)
+static bool
+input_rewind_available(struct input_stream *is)
{
struct input_rewind *r = (struct input_rewind *)is;
- int ret = input_stream_buffer(r->input, error_r);
- if (ret < 0 || !reading_from_buffer(r))
- copy_attributes(r);
-
- return ret;
+ return input_stream_available(r->input);
}
static size_t
@@ -232,7 +228,7 @@ static const struct input_plugin rewind_input_plugin = {
.check = input_rewind_check,
.update = input_rewind_update,
.tag = input_rewind_tag,
- .buffer = input_rewind_buffer,
+ .available = input_rewind_available,
.read = input_rewind_read,
.eof = input_rewind_eof,
.seek = input_rewind_seek,
@@ -251,7 +247,8 @@ input_rewind_open(struct input_stream *is)
return is;
c = g_new(struct input_rewind, 1);
- input_stream_init(&c->base, &rewind_input_plugin, is->uri);
+ input_stream_init(&c->base, &rewind_input_plugin, is->uri,
+ is->mutex, is->cond);
c->tail = 0;
c->input = is;
diff --git a/src/input/soup_input_plugin.c b/src/input/soup_input_plugin.c
index 23665c1a2..dc005a58c 100644
--- a/src/input/soup_input_plugin.c
+++ b/src/input/soup_input_plugin.c
@@ -46,9 +46,6 @@ static SoupSession *soup_session;
struct input_soup {
struct input_stream base;
- GMutex *mutex;
- GCond *cond;
-
SoupMessage *msg;
GQueue *buffers;
@@ -124,14 +121,14 @@ input_soup_session_callback(G_GNUC_UNUSED SoupSession *session,
assert(msg == s->msg);
assert(!s->completed);
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
s->base.ready = true;
s->alive = false;
s->completed = true;
- g_cond_broadcast(s->cond);
- g_mutex_unlock(s->mutex);
+ g_cond_broadcast(s->base.cond);
+ g_mutex_unlock(s->base.mutex);
}
static void
@@ -140,7 +137,7 @@ input_soup_got_headers(SoupMessage *msg, gpointer user_data)
struct input_soup *s = user_data;
if (!SOUP_STATUS_IS_SUCCESSFUL(msg->status_code)) {
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
if (s->postponed_error == NULL)
s->postponed_error =
@@ -148,7 +145,7 @@ input_soup_got_headers(SoupMessage *msg, gpointer user_data)
"got HTTP status %d",
msg->status_code);
- g_mutex_unlock(s->mutex);
+ g_mutex_unlock(s->base.mutex);
soup_session_cancel_message(soup_session, msg,
SOUP_STATUS_CANCELLED);
@@ -157,10 +154,10 @@ input_soup_got_headers(SoupMessage *msg, gpointer user_data)
soup_message_body_set_accumulate(msg->response_body, false);
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
s->base.ready = true;
- g_cond_broadcast(s->cond);
- g_mutex_unlock(s->mutex);
+ g_cond_broadcast(s->base.cond);
+ g_mutex_unlock(s->base.mutex);
}
static void
@@ -170,7 +167,7 @@ input_soup_got_chunk(SoupMessage *msg, SoupBuffer *chunk, gpointer user_data)
assert(msg == s->msg);
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
g_queue_push_tail(s->buffers, soup_buffer_copy(chunk));
s->total_buffered += chunk->length;
@@ -180,8 +177,8 @@ input_soup_got_chunk(SoupMessage *msg, SoupBuffer *chunk, gpointer user_data)
soup_session_pause_message(soup_session, msg);
}
- g_cond_broadcast(s->cond);
- g_mutex_unlock(s->mutex);
+ g_cond_broadcast(s->base.cond);
+ g_mutex_unlock(s->base.mutex);
}
static void
@@ -191,14 +188,14 @@ input_soup_got_body(G_GNUC_UNUSED SoupMessage *msg, gpointer user_data)
assert(msg == s->msg);
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
s->base.ready = true;
s->eof = true;
s->alive = false;
- g_cond_broadcast(s->cond);
- g_mutex_unlock(s->mutex);
+ g_cond_broadcast(s->base.cond);
+ g_mutex_unlock(s->base.mutex);
}
static bool
@@ -216,7 +213,7 @@ input_soup_wait_data(struct input_soup *s)
assert(s->current_consumed == 0);
- g_cond_wait(s->cond, s->mutex);
+ g_cond_wait(s->base.cond, s->base.mutex);
}
}
@@ -232,16 +229,16 @@ input_soup_queue(gpointer data)
}
static struct input_stream *
-input_soup_open(const char *uri, G_GNUC_UNUSED GError **error_r)
+input_soup_open(const char *uri,
+ GMutex *mutex, GCond *cond,
+ G_GNUC_UNUSED GError **error_r)
{
if (strncmp(uri, "http://", 7) != 0)
return NULL;
struct input_soup *s = g_new(struct input_soup, 1);
- input_stream_init(&s->base, &input_plugin_soup, uri);
-
- s->mutex = g_mutex_new();
- s->cond = g_cond_new();
+ input_stream_init(&s->base, &input_plugin_soup, uri,
+ mutex, cond);
s->buffers = g_queue_new();
s->current_consumed = 0;
@@ -288,25 +285,22 @@ input_soup_close(struct input_stream *is)
{
struct input_soup *s = (struct input_soup *)is;
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
if (!s->completed) {
/* the messages's session callback hasn't been invoked
yet; cancel it and wait for completion */
- g_mutex_unlock(s->mutex);
+ g_mutex_unlock(s->base.mutex);
io_thread_call(input_soup_cancel, s);
- g_mutex_lock(s->mutex);
+ g_mutex_lock(s->base.mutex);
while (!s->completed)
- g_cond_wait(s->cond, s->mutex);
+ g_cond_wait(s->base.cond, s->base.mutex);
}
- g_mutex_unlock(s->mutex);
-
- g_mutex_free(s->mutex);
- g_cond_free(s->cond);
+ g_mutex_unlock(s->base.mutex);
SoupBuffer *buffer;
while ((buffer = g_queue_pop_head(s->buffers)) != NULL)
@@ -325,54 +319,21 @@ input_soup_check(struct input_stream *is, GError **error_r)
{
struct input_soup *s = (struct input_soup *)is;
- g_mutex_lock(s->mutex);
-
bool success = s->postponed_error == NULL;
if (!success) {
g_propagate_error(error_r, s->postponed_error);
s->postponed_error = NULL;
}
- g_mutex_unlock(s->mutex);
return success;
}
-static int
-input_soup_buffer(struct input_stream *is, GError **error_r)
+static bool
+input_soup_available(struct input_stream *is)
{
struct input_soup *s = (struct input_soup *)is;
- g_mutex_lock(s->mutex);
-
- if (s->pause) {
- if (s->total_buffered >= SOUP_MAX_BUFFERED) {
- g_mutex_unlock(s->mutex);
- return 1;
- }
-
- s->pause = false;
- soup_session_unpause_message(soup_session, s->msg);
- }
-
-
- bool success = input_soup_wait_data(s);
-
- if (!success) {
- if (s->postponed_error != NULL) {
- g_propagate_error(error_r, s->postponed_error);
- s->postponed_error = NULL;
- } else
- g_set_error_literal(error_r, soup_quark(), 0,
- "HTTP failure");
- }
-
- g_mutex_unlock(s->mutex);
-
- if (!success) {
- return -1;
- }
-
- return 1;
+ return s->eof || !s->alive || !g_queue_is_empty(s->buffers);
}
static size_t
@@ -381,8 +342,6 @@ input_soup_read(struct input_stream *is, void *ptr, size_t size,
{
struct input_soup *s = (struct input_soup *)is;
- g_mutex_lock(s->mutex);
-
if (!input_soup_wait_data(s)) {
assert(!s->alive);
@@ -392,8 +351,6 @@ input_soup_read(struct input_stream *is, void *ptr, size_t size,
} else
g_set_error_literal(error_r, soup_quark(), 0,
"HTTP failure");
-
- g_mutex_unlock(s->mutex);
return 0;
}
@@ -442,7 +399,6 @@ input_soup_read(struct input_stream *is, void *ptr, size_t size,
size_t nbytes = p - p0;
s->base.offset += nbytes;
- g_mutex_unlock(s->mutex);
return nbytes;
}
@@ -462,7 +418,7 @@ const struct input_plugin input_plugin_soup = {
.open = input_soup_open,
.close = input_soup_close,
.check = input_soup_check,
- .buffer = input_soup_buffer,
+ .available = input_soup_available,
.read = input_soup_read,
.eof = input_soup_eof,
};