/*
* Copyright (C) 2003-2013 The Music Player Daemon Project
* http://www.musicpd.org
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License along
* with this program; if not, write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#include "config.h"
#include "CurlInputPlugin.hxx"
#include "InputStream.hxx"
#include "InputPlugin.hxx"
#include "ConfigGlobal.hxx"
#include "ConfigData.hxx"
#include "tag/Tag.hxx"
#include "IcyMetaDataParser.hxx"
#include "event/MultiSocketMonitor.hxx"
#include "event/Call.hxx"
#include "IOThread.hxx"
#include "util/ASCII.hxx"
#include "util/CharUtil.hxx"
#include "util/NumberParser.hxx"
#include "util/Error.hxx"
#include "util/Domain.hxx"
#include "Log.hxx"
#include <assert.h>
#if defined(WIN32)
#include <winsock2.h>
#else
#include <sys/select.h>
#endif
#include <string.h>
#include <errno.h>
#include <list>
#include <forward_list>
#include <curl/curl.h>
#include <glib.h>
#if LIBCURL_VERSION_NUM < 0x071200
#error libcurl is too old
#endif
/**
* 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;
/**
* Resume the stream at this number of bytes after it has been paused.
*/
static const size_t CURL_RESUME_AT = 384 * 1024;
/**
* Buffers created by input_curl_writefunction().
*/
class CurlInputBuffer {
/** size of the payload */
size_t size;
/** how much has been consumed yet? */
size_t consumed;
/** the payload */
uint8_t *data;
public:
CurlInputBuffer(const void *_data, size_t _size)
:size(_size), consumed(0), data(new uint8_t[size]) {
memcpy(data, _data, size);
}
~CurlInputBuffer() {
delete[] data;
}
CurlInputBuffer(const CurlInputBuffer &) = delete;
CurlInputBuffer &operator=(const CurlInputBuffer &) = delete;
const void *Begin() const {
return data + consumed;
}
size_t TotalSize() const {
return size;
}
size_t Available() const {
return size - consumed;
}
/**
* Mark a part of the buffer as consumed.
*
* @return false if the buffer is now empty
*/
bool Consume(size_t length) {
assert(consumed < size);
consumed += length;
if (consumed < size)
return true;
assert(consumed == size);
return false;
}
bool Read(void *dest, size_t length) {
assert(consumed + length <= size);
memcpy(dest, data + consumed, length);
return Consume(length);
}
};
struct input_curl {
InputStream base;
/* some buffers which were passed to libcurl, which we have
too free */
char range[32];
struct curl_slist *request_headers;
/** the curl handles */
CURL *easy;
/** list of buffers, where input_curl_writefunction() appends
to, and input_curl_read() reads from them */
std::list<CurlInputBuffer> buffers;
/**
* Is the connection currently paused? That happens when the
* buffer was getting too large. It will be unpaused when the
* buffer is below the threshold again.
*/
bool paused;
/** error message provided by libcurl */
char error[CURL_ERROR_SIZE];
/** parser for icy-metadata */
IcyMetaDataParser icy;
/** the stream name from the icy-name response header */
std::string meta_name;
/** the tag object ready to be requested via
InputStream::ReadTag() */
Tag *tag;
Error postponed_error;
input_curl(const char *url, Mutex &mutex, Cond &cond)
:base(input_plugin_curl, url, mutex, cond),
request_headers(nullptr),
paused(false),
tag(nullptr) {}
~input_curl();
input_curl(const input_curl &) = delete;
input_curl &operator=(const input_curl &) = delete;
};
/**
* This class monitors all CURL file descriptors.
*/
class CurlSockets final : private MultiSocketMonitor {
public:
CurlSockets(EventLoop &_loop)
:MultiSocketMonitor(_loop) {}
using MultiSocketMonitor::InvalidateSockets;
private:
void UpdateSockets();
virtual int PrepareSockets() override;
virtual void DispatchSockets() override;
};
/** libcurl should accept "ICY 200 OK" */
static struct curl_slist *http_200_aliases;
/** HTTP proxy settings */
static const char *proxy, *proxy_user, *proxy_password;
static unsigned proxy_port;
static struct {
CURLM *multi;
/**
* A linked list of all active HTTP requests. An active
* request is one that doesn't have the "eof" flag set.
*/
std::forward_list<input_curl *> requests;
CurlSockets *sockets;
} curl;
static constexpr Domain http_domain("http");
static constexpr Domain curl_domain("curl");
static constexpr Domain curlm_domain("curlm");
/**
* Find a request by its CURL "easy" handle.
*
* Runs in the I/O thread. No lock needed.
*/
static struct input_curl *
input_curl_find_request(CURL *easy)
{
assert(io_thread_inside());
for (auto c : curl.requests)
if (c->easy == easy)
return c;
return nullptr;
}
static void
input_curl_resume(struct input_curl *c)
{
assert(io_thread_inside());
if (c->paused) {
c->paused = false;
curl_easy_pause(c->easy, CURLPAUSE_CONT);
curl.sockets->InvalidateSockets();
}
}
/**
* Calculates the GLib event bit mask for one file descriptor,
* obtained from three #fd_set objects filled by curl_multi_fdset().
*/
static unsigned
input_curl_fd_events(int fd, fd_set *rfds, fd_set *wfds, fd_set *efds)
{
unsigned events = 0;
if (FD_ISSET(fd, rfds)) {
events |= MultiSocketMonitor::READ | MultiSocketMonitor::HANGUP
| MultiSocketMonitor::ERROR;
FD_CLR(fd, rfds);
}
if (FD_ISSET(fd, wfds)) {
events |= MultiSocketMonitor::WRITE |
MultiSocketMonitor::ERROR;
FD_CLR(fd, wfds);
}
if (FD_ISSET(fd, efds)) {
events |= MultiSocketMonitor::HANGUP |
MultiSocketMonitor::ERROR;
FD_CLR(fd, efds);
}
return events;
}
/**
* Updates all registered GPollFD objects, unregisters old ones,
* registers new ones.
*
* Runs in the I/O thread. No lock needed.
*/
void
CurlSockets::UpdateSockets()
{
assert(io_thread_inside());
fd_set rfds, wfds, efds;
FD_ZERO(&rfds);
FD_ZERO(&wfds);
FD_ZERO(&efds);
int max_fd;
CURLMcode mcode = curl_multi_fdset(curl.multi, &rfds, &wfds,
&efds, &max_fd);
if (mcode != CURLM_OK) {
FormatError(curlm_domain,
"curl_multi_fdset() failed: %s",
curl_multi_strerror(mcode));
return;
}
UpdateSocketList([&rfds, &wfds, &efds](int fd){
return input_curl_fd_events(fd, &rfds,
&wfds, &efds);
});
for (int fd = 0; fd <= max_fd; ++fd) {
unsigned events = input_curl_fd_events(fd, &rfds, &wfds, &efds);
if (events != 0)
AddSocket(fd, events);
}
}
/**
* Runs in the I/O thread. No lock needed.
*/
static bool
input_curl_easy_add(struct input_curl *c, Error &error)
{
assert(io_thread_inside());
assert(c != nullptr);
assert(c->easy != nullptr);
assert(input_curl_find_request(c->easy) == nullptr);
curl.requests.push_front(c);
CURLMcode mcode = curl_multi_add_handle(curl.multi, c->easy);
if (mcode != CURLM_OK) {
error.Format(curlm_domain, mcode,
"curl_multi_add_handle() failed: %s",
curl_multi_strerror(mcode));
return false;
}
curl.sockets->InvalidateSockets();
return true;
}
/**
* Call input_curl_easy_add() in the I/O thread. May be called from
* any thread. Caller must not hold a mutex.
*/
static bool
input_curl_easy_add_indirect(struct input_curl *c, Error &error)
{
assert(c != nullptr);
assert(c->easy != nullptr);
bool result;
BlockingCall(io_thread_get(), [c, &error, &result](){
result = input_curl_easy_add(c, error);
});
return result;
}
/**
* Frees the current "libcurl easy" handle, and everything associated
* with it.
*
* Runs in the I/O thread.
*/
static void
input_curl_easy_free(struct input_curl *c)
{
assert(io_thread_inside());
assert(c != nullptr);
if (c->easy == nullptr)
return;
curl.requests.remove(c);
curl_multi_remove_handle(curl.multi, c->easy);
curl_easy_cleanup(c->easy);
c->easy = nullptr;
curl_slist_free_all(c->request_headers);
c->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(struct input_curl *c)
{
BlockingCall(io_thread_get(), [c](){
input_curl_easy_free(c);
curl.sockets->InvalidateSockets();
});
assert(c->easy == nullptr);
}
/**
* Abort and free all HTTP requests.
*
* Runs in the I/O thread. The caller must not hold locks.
*/
static void
input_curl_abort_all_requests(const Error &error)
{
assert(io_thread_inside());
assert(error.IsDefined());
while (!curl.requests.empty()) {
struct input_curl *c = curl.requests.front();
assert(!c->postponed_error.IsDefined());
input_curl_easy_free(c);
const ScopeLock protect(c->base.mutex);
c->postponed_error.Set(error);
c->base.ready = true;
c->base.cond.broadcast();
}
}
/**
* A HTTP request is finished.
*
* Runs in the I/O thread. The caller must not hold locks.
*/
static void
input_curl_request_done(struct input_curl *c, CURLcode result, long status)
{
assert(io_thread_inside());
assert(c != nullptr);
assert(c->easy == nullptr);
assert(!c->postponed_error.IsDefined());
const ScopeLock protect(c->base.mutex);
if (result != CURLE_OK) {
c->postponed_error.Format(curl_domain, result,
"curl failed: %s", c->error);
} else if (status < 200 || status >= 300) {
c->postponed_error.Format(http_domain, status,
"got HTTP status %ld",
status);
}
c->base.ready = true;
c->base.cond.broadcast();
}
static void
input_curl_handle_done(CURL *easy_handle, CURLcode result)
{
struct input_curl *c = input_curl_find_request(easy_handle);
assert(c != nullptr);
long status = 0;
curl_easy_getinfo(easy_handle, CURLINFO_RESPONSE_CODE, &status);
input_curl_easy_free(c);
input_curl_request_done(c, result, status);
}
/**
* Check for finished HTTP responses.
*
* Runs in the I/O thread. The caller must not hold locks.
*/
static void
input_curl_info_read(void)
{
assert(io_thread_inside());
CURLMsg *msg;
int msgs_in_queue;
while ((msg = curl_multi_info_read(curl.multi,
&msgs_in_queue)) != nullptr) {
if (msg->msg == CURLMSG_DONE)
input_curl_handle_done(msg->easy_handle, msg->data.result);
}
}
/**
* Give control to CURL.
*
* Runs in the I/O thread. The caller must not hold locks.
*/
static bool
input_curl_perform(void)
{
assert(io_thread_inside());
CURLMcode mcode;
do {
int running_handles;
mcode = curl_multi_perform(curl.multi, &running_handles);
} while (mcode == CURLM_CALL_MULTI_PERFORM);
if (mcode != CURLM_OK && mcode != CURLM_CALL_MULTI_PERFORM) {
Error error;
error.Format(curlm_domain, mcode,
"curl_multi_perform() failed: %s",
curl_multi_strerror(mcode));
input_curl_abort_all_requests(error);
return false;
}
return true;
}
int
CurlSockets::PrepareSockets()
{
UpdateSockets();
long timeout2;
CURLMcode mcode = curl_multi_timeout(curl.multi, &timeout2);
if (mcode == CURLM_OK) {
if (timeout2 >= 0 && timeout2 < 10)
/* CURL 7.21.1 likes to report "timeout=0",
which means we're running in a busy loop.
Quite a bad idea to waste so much CPU.
Let's use a lower limit of 10ms. */
timeout2 = 10;
return timeout2;
} else {
FormatWarning(curlm_domain,
"curl_multi_timeout() failed: %s",
curl_multi_strerror(mcode));
return -1;
}
}
void
CurlSockets::DispatchSockets()
{
if (input_curl_perform())
input_curl_info_read();
}
/*
* InputPlugin methods
*
*/
static bool
input_curl_init(const config_param ¶m, Error &error)
{
CURLcode code = curl_global_init(CURL_GLOBAL_ALL);
if (code != CURLE_OK) {
error.Format(curl_domain, code,
"curl_global_init() failed: %s",
curl_easy_strerror(code));
return false;
}
http_200_aliases = curl_slist_append(http_200_aliases, "ICY 200 OK");
proxy = param.GetBlockValue("proxy");
proxy_port = param.GetBlockValue("proxy_port", 0u);
proxy_user = param.GetBlockValue("proxy_user");
proxy_password = param.GetBlockValue("proxy_password");
if (proxy == nullptr) {
/* deprecated proxy configuration */
proxy = config_get_string(CONF_HTTP_PROXY_HOST, nullptr);
proxy_port = config_get_positive(CONF_HTTP_PROXY_PORT, 0);
proxy_user = config_get_string(CONF_HTTP_PROXY_USER, nullptr);
proxy_password = config_get_string(CONF_HTTP_PROXY_PASSWORD,
"");
}
curl.multi = curl_multi_init();
if (curl.multi == nullptr) {
error.Set(curl_domain, 0, "curl_multi_init() failed");
return false;
}
curl.sockets = new CurlSockets(io_thread_get());
return true;
}
static void
input_curl_finish(void)
{
assert(curl.requests.empty());
BlockingCall(io_thread_get(), [](){
delete curl.sockets;
});
curl_multi_cleanup(curl.multi);
curl_slist_free_all(http_200_aliases);
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 struct input_curl *c)
{
size_t total = 0;
for (const auto &i : c->buffers)
total += i.TotalSize();
return total;
}
input_curl::~input_curl()
{
delete tag;
input_curl_easy_free_indirect(this);
}
static bool
input_curl_check(InputStream *is, Error &error)
{
struct input_curl *c = (struct input_curl *)is;
bool success = !c->postponed_error.IsDefined();
if (!success) {
error = std::move(c->postponed_error);
c->postponed_error.Clear();
}
return success;
}
static Tag *
input_curl_tag(InputStream *is)
{
struct input_curl *c = (struct input_curl *)is;
Tag *tag = c->tag;
c->tag = nullptr;
return tag;
}
static bool
fill_buffer(struct input_curl *c, Error &error)
{
while (c->easy != nullptr && c->buffers.empty())
c->base.cond.wait(c->base.mutex);
if (c->postponed_error.IsDefined()) {
error = std::move(c->postponed_error);
c->postponed_error.Clear();
return false;
}
return !c->buffers.empty();
}
static size_t
read_from_buffer(IcyMetaDataParser &icy, std::list<CurlInputBuffer> &buffers,
void *dest0, size_t length)
{
auto &buffer = buffers.front();
uint8_t *dest = (uint8_t *)dest0;
size_t nbytes = 0;
if (length > buffer.Available())
length = buffer.Available();
while (true) {
size_t chunk;
chunk = icy.Data(length);
if (chunk > 0) {
const bool empty = !buffer.Read(dest, chunk);
nbytes += chunk;
dest += chunk;
length -= chunk;
if (empty) {
buffers.pop_front();
break;
}
if (length == 0)
break;
}
chunk = icy.Meta(buffer.Begin(), length);
if (chunk > 0) {
const bool empty = !buffer.Consume(chunk);
length -= chunk;
if (empty) {
buffers.pop_front();
break;
}
if (length == 0)
break;
}
}
return nbytes;
}
static void
copy_icy_tag(struct input_curl *c)
{
Tag *tag = c->icy.ReadTag();
if (tag == nullptr)
return;
delete c->tag;
if (!c->meta_name.empty() && !tag->HasType(TAG_NAME))
tag->AddItem(TAG_NAME, c->meta_name.c_str());
c->tag = tag;
}
static bool
input_curl_available(InputStream *is)
{
struct input_curl *c = (struct input_curl *)is;
return c->postponed_error.IsDefined() || c->easy == nullptr ||
!c->buffers.empty();
}
static size_t
input_curl_read(InputStream *is, void *ptr, size_t size,
Error &error)
{
struct input_curl *c = (struct input_curl *)is;
bool success;
size_t nbytes = 0;
char *dest = (char *)ptr;
do {
/* fill the buffer */
success = fill_buffer(c, error);
if (!success)
return 0;
/* send buffer contents */
while (size > 0 && !c->buffers.empty()) {
size_t copy = read_from_buffer(c->icy, c->buffers,
dest + nbytes, size);
nbytes += copy;
size -= copy;
}
} while (nbytes == 0);
if (c->icy.IsDefined())
copy_icy_tag(c);
is->offset += (InputPlugin::offset_type)nbytes;
if (c->paused && curl_total_buffer_size(c) < CURL_RESUME_AT) {
c->base.mutex.unlock();
BlockingCall(io_thread_get(), [c](){
input_curl_resume(c);
});
c->base.mutex.lock();
}
return nbytes;
}
static void
input_curl_close(InputStream *is)
{
struct input_curl *c = (struct input_curl *)is;
delete c;
}
static bool
input_curl_eof(gcc_unused InputStream *is)
{
struct input_curl *c = (struct input_curl *)is;
return c->easy == nullptr && c->buffers.empty();
}
/** called by curl when new data is available */
static size_t
input_curl_headerfunction(void *ptr, size_t size, size_t nmemb, void *stream)
{
struct input_curl *c = (struct input_curl *)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;
} else if (StringEqualsCaseASCII(name, "content-length")) {
char buffer[64];
if ((size_t)(end - header) >= sizeof(buffer))
return size;
memcpy(buffer, value, end - value);
buffer[end - value] = 0;
c->base.size = c->base.offset + ParseUint64(buffer);
} else if (StringEqualsCaseASCII(name, "content-type")) {
c->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);
delete c->tag;
c->tag = new Tag();
c->tag->AddItem(TAG_NAME, c->meta_name.c_str());
} 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;
memcpy(buffer, value, end - value);
buffer[end - value] = 0;
icy_metaint = ParseUint64(buffer);
FormatDebug(curl_domain, "icy-metaint=%zu", icy_metaint);
if (icy_metaint > 0) {
c->icy.Start(icy_metaint);
/* a stream with icy-metadata is not
seekable */
c->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)
{
struct input_curl *c = (struct input_curl *)stream;
size *= nmemb;
if (size == 0)
return 0;
const ScopeLock protect(c->base.mutex);
if (curl_total_buffer_size(c) + size >= CURL_MAX_BUFFERED) {
c->paused = true;
return CURL_WRITEFUNC_PAUSE;
}
c->buffers.emplace_back(ptr, size);
c->base.ready = true;
c->base.cond.broadcast();
return size;
}
static bool
input_curl_easy_init(struct input_curl *c, Error &error)
{
CURLcode code;
c->easy = curl_easy_init();
if (c->easy == nullptr) {
error.Set(curl_domain, "curl_easy_init() failed");
return false;
}
curl_easy_setopt(c->easy, CURLOPT_USERAGENT,
"Music Player Daemon " VERSION);
curl_easy_setopt(c->easy, CURLOPT_HEADERFUNCTION,
input_curl_headerfunction);
curl_easy_setopt(c->easy, CURLOPT_WRITEHEADER, c);
curl_easy_setopt(c->easy, CURLOPT_WRITEFUNCTION,
input_curl_writefunction);
curl_easy_setopt(c->easy, CURLOPT_WRITEDATA, c);
curl_easy_setopt(c->easy, CURLOPT_HTTP200ALIASES, http_200_aliases);
curl_easy_setopt(c->easy, CURLOPT_FOLLOWLOCATION, 1);
curl_easy_setopt(c->easy, CURLOPT_NETRC, 1);
curl_easy_setopt(c->easy, CURLOPT_MAXREDIRS, 5);
curl_easy_setopt(c->easy, CURLOPT_FAILONERROR, true);
curl_easy_setopt(c->easy, CURLOPT_ERRORBUFFER, c->error);
curl_easy_setopt(c->easy, CURLOPT_NOPROGRESS, 1l);
curl_easy_setopt(c->easy, CURLOPT_NOSIGNAL, 1l);
curl_easy_setopt(c->easy, CURLOPT_CONNECTTIMEOUT, 10l);
if (proxy != nullptr)
curl_easy_setopt(c->easy, CURLOPT_PROXY, proxy);
if (proxy_port > 0)
curl_easy_setopt(c->easy, CURLOPT_PROXYPORT, (long)proxy_port);
if (proxy_user != nullptr && proxy_password != nullptr) {
char proxy_auth_str[1024];
snprintf(proxy_auth_str, sizeof(proxy_auth_str),
"%s:%s",
proxy_user, proxy_password);
curl_easy_setopt(c->easy, CURLOPT_PROXYUSERPWD, proxy_auth_str);
}
code = curl_easy_setopt(c->easy, CURLOPT_URL, c->base.uri.c_str());
if (code != CURLE_OK) {
error.Format(curl_domain, code,
"curl_easy_setopt() failed: %s",
curl_easy_strerror(code));
return false;
}
c->request_headers = nullptr;
c->request_headers = curl_slist_append(c->request_headers,
"Icy-Metadata: 1");
curl_easy_setopt(c->easy, CURLOPT_HTTPHEADER, c->request_headers);
return true;
}
static bool
input_curl_seek(InputStream *is, InputPlugin::offset_type offset,
int whence,
Error &error)
{
struct input_curl *c = (struct input_curl *)is;
bool ret;
assert(is->ready);
if (whence == SEEK_SET && offset == is->offset)
/* no-op */
return true;
if (!is->seekable)
return false;
/* calculate the absolute offset */
switch (whence) {
case SEEK_SET:
break;
case SEEK_CUR:
offset += is->offset;
break;
case SEEK_END:
if (is->size < 0)
/* stream size is not known */
return false;
offset += is->size;
break;
default:
return false;
}
if (offset < 0)
return false;
/* check if we can fast-forward the buffer */
while (offset > is->offset && !c->buffers.empty()) {
auto &buffer = c->buffers.front();
size_t length = buffer.Available();
if (offset - is->offset < (InputPlugin::offset_type)length)
length = offset - is->offset;
const bool empty = !buffer.Consume(length);
if (empty)
c->buffers.pop_front();
is->offset += length;
}
if (offset == is->offset)
return true;
/* close the old connection and open a new one */
c->base.mutex.unlock();
input_curl_easy_free_indirect(c);
c->buffers.clear();
is->offset = offset;
if (is->offset == is->size) {
/* seek to EOF: simulate empty result; avoid
triggering a "416 Requested Range Not Satisfiable"
response */
return true;
}
ret = input_curl_easy_init(c, error);
if (!ret)
return false;
/* send the "Range" header */
if (is->offset > 0) {
sprintf(c->range, "%lld-", (long long)is->offset);
curl_easy_setopt(c->easy, CURLOPT_RANGE, c->range);
}
c->base.ready = false;
if (!input_curl_easy_add_indirect(c, error))
return false;
c->base.mutex.lock();
while (!c->base.ready)
c->base.cond.wait(c->base.mutex);
if (c->postponed_error.IsDefined()) {
error = std::move(c->postponed_error);
c->postponed_error.Clear();
return false;
}
return true;
}
static InputStream *
input_curl_open(const char *url, Mutex &mutex, Cond &cond,
Error &error)
{
if (memcmp(url, "http://", 7) != 0 &&
memcmp(url, "https://", 8) != 0)
return nullptr;
struct input_curl *c = new input_curl(url, mutex, cond);
if (!input_curl_easy_init(c, error)) {
delete c;
return nullptr;
}
if (!input_curl_easy_add_indirect(c, error)) {
delete c;
return nullptr;
}
return &c->base;
}
const struct InputPlugin input_plugin_curl = {
"curl",
input_curl_init,
input_curl_finish,
input_curl_open,
input_curl_close,
input_curl_check,
nullptr,
input_curl_tag,
input_curl_available,
input_curl_read,
input_curl_eof,
input_curl_seek,
};