diff options
Diffstat (limited to 'src/input')
| -rw-r--r-- | src/input/archive_input_plugin.c | 7 | ||||
| -rw-r--r-- | src/input/cdio_paranoia_input_plugin.c | 7 | ||||
| -rw-r--r-- | src/input/curl_input_plugin.c | 108 | ||||
| -rw-r--r-- | src/input/despotify_input_plugin.c | 7 | ||||
| -rw-r--r-- | src/input/ffmpeg_input_plugin.c | 7 | ||||
| -rw-r--r-- | src/input/file_input_plugin.c | 7 | ||||
| -rw-r--r-- | src/input/mms_input_plugin.c | 7 | ||||
| -rw-r--r-- | src/input/rewind_input_plugin.c | 15 | ||||
| -rw-r--r-- | src/input/soup_input_plugin.c | 102 |
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, }; |