diff options
| -rw-r--r-- | src/input/plugins/CurlInputPlugin.cxx | 409 |
1 files changed, 231 insertions, 178 deletions
diff --git a/src/input/plugins/CurlInputPlugin.cxx b/src/input/plugins/CurlInputPlugin.cxx index bc4b0e5b6..a339353cd 100644 --- a/src/input/plugins/CurlInputPlugin.cxx +++ b/src/input/plugins/CurlInputPlugin.cxx @@ -177,6 +177,63 @@ struct CurlInputStream { CurlInputStream(const CurlInputStream &) = delete; CurlInputStream &operator=(const CurlInputStream &) = delete; + + bool Check(Error &error); + + bool IsEOF() const { + return easy == nullptr && buffers.empty(); + } + + Tag *ReadTag(); + + bool IsAvailable() const { + return postponed_error.IsDefined() || easy == nullptr || + !buffers.empty(); + } + + size_t Read(void *ptr, size_t size, Error &error); + + /** + * Frees the current "libcurl easy" handle, and everything + * associated with it. + * + * Runs in the I/O thread. + */ + void FreeEasy(); + + /** + * Frees the current "libcurl easy" handle, and everything associated + * with it. + * + * The mutex must not be locked. + */ + void FreeEasyIndirect(); + + void HeaderReceived(const char *name, + const char *value, const char *end); + + size_t DataReceived(const void *ptr, size_t size); + + void Resume(); + bool FillBuffer(Error &error); + + /** + * Determine the total sizes of all buffers, including + * portions that have already been consumed. + * + * The caller must lock the mutex. + */ + gcc_pure + size_t GetTotalBufferSize() const; + + void CopyIcyTag(); + + /** + * A HTTP request is finished. + * + * Runs in the I/O thread. The caller must not hold locks. + */ + void RequestDone(CURLcode result, long status); }; class CurlMulti; @@ -335,14 +392,14 @@ input_curl_find_request(CURL *easy) return (CurlInputStream *)p; } -static void -input_curl_resume(CurlInputStream *c) +inline void +CurlInputStream::Resume() { assert(io_thread_inside()); - if (c->paused) { - c->paused = false; - curl_easy_pause(c->easy, CURLPAUSE_CONT); + if (paused) { + paused = false; + curl_easy_pause(easy, CURLPAUSE_CONT); if (curl_version_num < 0x072000) /* libcurl older than 7.32.0 does not update @@ -445,74 +502,55 @@ CurlMulti::Remove(CurlInputStream *c) curl_multi_remove_handle(multi, c->easy); } -/** - * Frees the current "libcurl easy" handle, and everything associated - * with it. - * - * Runs in the I/O thread. - */ -static void -input_curl_easy_free(CurlInputStream *c) +void +CurlInputStream::FreeEasy() { assert(io_thread_inside()); - assert(c != nullptr); - if (c->easy == nullptr) + if (easy == nullptr) return; - curl_multi->Remove(c); + curl_multi->Remove(this); - curl_easy_cleanup(c->easy); - c->easy = nullptr; + curl_easy_cleanup(easy); + easy = nullptr; - curl_slist_free_all(c->request_headers); - c->request_headers = nullptr; + curl_slist_free_all(request_headers); + request_headers = nullptr; } -/** - * Frees the current "libcurl easy" handle, and everything associated - * with it. - * - * The mutex must not be locked. - */ -static void -input_curl_easy_free_indirect(CurlInputStream *c) +void +CurlInputStream::FreeEasyIndirect() { - BlockingCall(io_thread_get(), [c](){ - input_curl_easy_free(c); + BlockingCall(io_thread_get(), [this](){ + FreeEasy(); curl_multi->InvalidateSockets(); }); - assert(c->easy == nullptr); + assert(easy == nullptr); } -/** - * A HTTP request is finished. - * - * Runs in the I/O thread. The caller must not hold locks. - */ -static void -input_curl_request_done(CurlInputStream *c, CURLcode result, long status) +inline void +CurlInputStream::RequestDone(CURLcode result, long status) { assert(io_thread_inside()); - assert(c != nullptr); - assert(c->easy == nullptr); - assert(!c->postponed_error.IsDefined()); + assert(!postponed_error.IsDefined()); + + FreeEasy(); - const ScopeLock protect(c->base.mutex); + const ScopeLock protect(base.mutex); if (result != CURLE_OK) { - c->postponed_error.Format(curl_domain, result, - "curl failed: %s", c->error_buffer); + postponed_error.Format(curl_domain, result, + "curl failed: %s", error_buffer); } else if (status < 200 || status >= 300) { - c->postponed_error.Format(http_domain, status, - "got HTTP status %ld", - status); + postponed_error.Format(http_domain, status, + "got HTTP status %ld", + status); } - c->base.ready = true; - - c->base.cond.broadcast(); + base.ready = true; + base.cond.broadcast(); } static void @@ -524,8 +562,7 @@ input_curl_handle_done(CURL *easy_handle, CURLcode result) long status = 0; curl_easy_getinfo(easy_handle, CURLINFO_RESPONSE_CODE, &status); - input_curl_easy_free(c); - input_curl_request_done(c, result, status); + c->RequestDone(result, status); } void @@ -656,19 +693,11 @@ input_curl_finish(void) curl_global_cleanup(); } -/** - * Determine the total sizes of all buffers, including portions that - * have already been consumed. - * - * The caller must lock the mutex. - */ -gcc_pure -static size_t -curl_total_buffer_size(const CurlInputStream *c) +size_t +CurlInputStream::GetTotalBufferSize() const { size_t total = 0; - - for (const auto &i : c->buffers) + for (const auto &i : buffers) total += i.TotalSize(); return total; @@ -678,46 +707,56 @@ CurlInputStream::~CurlInputStream() { delete tag; - input_curl_easy_free_indirect(this); + FreeEasyIndirect(); } -static bool -input_curl_check(InputStream *is, Error &error) +inline bool +CurlInputStream::Check(Error &error) { - CurlInputStream *c = (CurlInputStream *)is; - - bool success = !c->postponed_error.IsDefined(); + bool success = !postponed_error.IsDefined(); if (!success) { - error = std::move(c->postponed_error); - c->postponed_error.Clear(); + error = std::move(postponed_error); + postponed_error.Clear(); } return success; } +static bool +input_curl_check(InputStream *is, Error &error) +{ + CurlInputStream &c = *(CurlInputStream *)is; + return c.Check(error); +} + +inline Tag * +CurlInputStream::ReadTag() +{ + Tag *result = tag; + tag = nullptr; + return result; +} + static Tag * input_curl_tag(InputStream *is) { - CurlInputStream *c = (CurlInputStream *)is; - Tag *tag = c->tag; - - c->tag = nullptr; - return tag; + CurlInputStream &c = *(CurlInputStream *)is; + return c.ReadTag(); } -static bool -fill_buffer(CurlInputStream *c, Error &error) +inline bool +CurlInputStream::FillBuffer(Error &error) { - while (c->easy != nullptr && c->buffers.empty()) - c->base.cond.wait(c->base.mutex); + while (easy != nullptr && buffers.empty()) + base.cond.wait(base.mutex); - if (c->postponed_error.IsDefined()) { - error = std::move(c->postponed_error); - c->postponed_error.Clear(); + if (postponed_error.IsDefined()) { + error = std::move(postponed_error); + postponed_error.Clear(); return false; } - return !c->buffers.empty(); + return !buffers.empty(); } static size_t @@ -770,54 +809,47 @@ read_from_buffer(IcyMetaDataParser &icy, std::list<CurlInputBuffer> &buffers, return nbytes; } -static void -copy_icy_tag(CurlInputStream *c) +inline void +CurlInputStream::CopyIcyTag() { - Tag *tag = c->icy.ReadTag(); - - if (tag == nullptr) + Tag *new_tag = icy.ReadTag(); + if (new_tag == nullptr) return; - delete c->tag; + delete tag; - if (!c->meta_name.empty() && !tag->HasType(TAG_NAME)) { - TagBuilder tag_builder(std::move(*tag)); - tag_builder.AddItem(TAG_NAME, c->meta_name.c_str()); - *tag = tag_builder.Commit(); + if (!meta_name.empty() && !new_tag->HasType(TAG_NAME)) { + TagBuilder tag_builder(std::move(*new_tag)); + tag_builder.AddItem(TAG_NAME, meta_name.c_str()); + *new_tag = tag_builder.Commit(); } - c->tag = tag; + tag = new_tag; } static bool input_curl_available(InputStream *is) { - CurlInputStream *c = (CurlInputStream *)is; - - return c->postponed_error.IsDefined() || c->easy == nullptr || - !c->buffers.empty(); + const CurlInputStream &c = *(const CurlInputStream *)is; + return c.IsAvailable(); } -static size_t -input_curl_read(InputStream *is, void *ptr, size_t size, - Error &error) +inline size_t +CurlInputStream::Read(void *ptr, size_t size, Error &error) { - CurlInputStream *c = (CurlInputStream *)is; - bool success; size_t nbytes = 0; char *dest = (char *)ptr; do { /* fill the buffer */ - success = fill_buffer(c, error); - if (!success) + if (!FillBuffer(error)) return 0; /* send buffer contents */ - while (size > 0 && !c->buffers.empty()) { - size_t copy = read_from_buffer(c->icy, c->buffers, + while (size > 0 && !buffers.empty()) { + size_t copy = read_from_buffer(icy, buffers, dest + nbytes, size); nbytes += copy; @@ -825,24 +857,32 @@ input_curl_read(InputStream *is, void *ptr, size_t size, } } while (nbytes == 0); - if (c->icy.IsDefined()) - copy_icy_tag(c); + if (icy.IsDefined()) + CopyIcyTag(); - is->offset += (InputPlugin::offset_type)nbytes; + base.offset += (InputPlugin::offset_type)nbytes; - if (c->paused && curl_total_buffer_size(c) < CURL_RESUME_AT) { - c->base.mutex.unlock(); + if (paused && GetTotalBufferSize() < CURL_RESUME_AT) { + base.mutex.unlock(); - BlockingCall(io_thread_get(), [c](){ - input_curl_resume(c); + BlockingCall(io_thread_get(), [this](){ + Resume(); }); - c->base.mutex.lock(); + base.mutex.lock(); } return nbytes; } +static size_t +input_curl_read(InputStream *is, void *ptr, size_t size, + Error &error) +{ + CurlInputStream &c = *(CurlInputStream *)is; + return c.Read(ptr, size, error); +} + static void input_curl_close(InputStream *is) { @@ -854,76 +894,48 @@ input_curl_close(InputStream *is) static bool input_curl_eof(gcc_unused InputStream *is) { - CurlInputStream *c = (CurlInputStream *)is; - - return c->easy == nullptr && c->buffers.empty(); + const CurlInputStream &c = *(const CurlInputStream *)is; + return c.IsEOF(); } -/** called by curl when new data is available */ -static size_t -input_curl_headerfunction(void *ptr, size_t size, size_t nmemb, void *stream) +inline void +CurlInputStream::HeaderReceived(const char *name, + const char *value, const char *end) { - CurlInputStream *c = (CurlInputStream *)stream; - char name[64]; - - size *= nmemb; - - const char *header = (const char *)ptr; - const char *end = header + size; - - const char *value = (const char *)memchr(header, ':', size); - if (value == nullptr || (size_t)(value - header) >= sizeof(name)) - return size; - - memcpy(name, header, value - header); - name[value - header] = 0; - - /* skip the colon */ - - ++value; - - /* strip the value */ - - while (value < end && IsWhitespaceOrNull(*value)) - ++value; - - while (end > value && IsWhitespaceOrNull(end[-1])) - --end; - if (StringEqualsCaseASCII(name, "accept-ranges")) { /* a stream with icy-metadata is not seekable */ - if (!c->icy.IsDefined()) - c->base.seekable = true; + if (!icy.IsDefined()) + base.seekable = true; } else if (StringEqualsCaseASCII(name, "content-length")) { char buffer[64]; - if ((size_t)(end - header) >= sizeof(buffer)) - return size; + if ((size_t)(end - value) >= sizeof(buffer)) + return; memcpy(buffer, value, end - value); buffer[end - value] = 0; - c->base.size = c->base.offset + ParseUint64(buffer); + base.size = base.offset + ParseUint64(buffer); } else if (StringEqualsCaseASCII(name, "content-type")) { - c->base.mime.assign(value, end); + base.mime.assign(value, end); } else if (StringEqualsCaseASCII(name, "icy-name") || StringEqualsCaseASCII(name, "ice-name") || StringEqualsCaseASCII(name, "x-audiocast-name")) { - c->meta_name.assign(value, end); + meta_name.assign(value, end); - delete c->tag; + delete tag; TagBuilder tag_builder; - tag_builder.AddItem(TAG_NAME, c->meta_name.c_str()); + tag_builder.AddItem(TAG_NAME, meta_name.c_str()); - c->tag = tag_builder.CommitNew(); + tag = tag_builder.CommitNew(); } else if (StringEqualsCaseASCII(name, "icy-metaint")) { char buffer[64]; size_t icy_metaint; - if ((size_t)(end - header) >= sizeof(buffer) || - c->icy.IsDefined()) - return size; + if ((size_t)(end - value) >= sizeof(buffer) || + icy.IsDefined()) + return; memcpy(buffer, value, end - value); buffer[end - value] = 0; @@ -932,41 +944,82 @@ input_curl_headerfunction(void *ptr, size_t size, size_t nmemb, void *stream) FormatDebug(curl_domain, "icy-metaint=%zu", icy_metaint); if (icy_metaint > 0) { - c->icy.Start(icy_metaint); + icy.Start(icy_metaint); /* a stream with icy-metadata is not seekable */ - c->base.seekable = false; + base.seekable = false; } } - - return size; } /** called by curl when new data is available */ static size_t -input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream) +input_curl_headerfunction(void *ptr, size_t size, size_t nmemb, void *stream) { - CurlInputStream *c = (CurlInputStream *)stream; + CurlInputStream &c = *(CurlInputStream *)stream; size *= nmemb; - if (size == 0) - return 0; - const ScopeLock protect(c->base.mutex); + const char *header = (const char *)ptr; + const char *end = header + size; + + char name[64]; + + const char *value = (const char *)memchr(header, ':', size); + if (value == nullptr || (size_t)(value - header) >= sizeof(name)) + return size; + + memcpy(name, header, value - header); + name[value - header] = 0; + + /* skip the colon */ + + ++value; + + /* strip the value */ + + while (value < end && IsWhitespaceOrNull(*value)) + ++value; + + while (end > value && IsWhitespaceOrNull(end[-1])) + --end; - if (curl_total_buffer_size(c) + size >= CURL_MAX_BUFFERED) { - c->paused = true; + c.HeaderReceived(name, value, end); + return size; +} + +inline size_t +CurlInputStream::DataReceived(const void *ptr, size_t size) +{ + assert(size > 0); + + const ScopeLock protect(base.mutex); + + if (GetTotalBufferSize() + size >= CURL_MAX_BUFFERED) { + paused = true; return CURL_WRITEFUNC_PAUSE; } - c->buffers.emplace_back(ptr, size); - c->base.ready = true; - - c->base.cond.broadcast(); + buffers.emplace_back(ptr, size); + base.ready = true; + base.cond.broadcast(); return size; } +/** called by curl when new data is available */ +static size_t +input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream) +{ + CurlInputStream &c = *(CurlInputStream *)stream; + + size *= nmemb; + if (size == 0) + return 0; + + return c.DataReceived(ptr, size); +} + static bool input_curl_easy_init(CurlInputStream *c, Error &error) { @@ -1091,7 +1144,7 @@ input_curl_seek(InputStream *is, InputPlugin::offset_type offset, c->base.mutex.unlock(); - input_curl_easy_free_indirect(c); + c->FreeEasyIndirect(); c->buffers.clear(); is->offset = offset; |