src/iohandler/curl_io_handler.cc
Rama master · réplica de lectura del repositorio original.
/*MT*
MediaTomb - http://www.mediatomb.cc/
curl_io_handler.cc - this file is part of MediaTomb.
Copyright (C) 2005 Gena Batyan <bgeradz@mediatomb.cc>,
Sergey 'Jin' Bostandzhyan <jin@mediatomb.cc>
Copyright (C) 2006-2010 Gena Batyan <bgeradz@mediatomb.cc>,
Sergey 'Jin' Bostandzhyan <jin@mediatomb.cc>,
Leonhard Wimmer <leo@mediatomb.cc>
Copyright (C) 2016-2026 Gerbera Contributors
MediaTomb is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License version 2
as published by the Free Software Foundation.
MediaTomb 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
version 2 along with MediaTomb; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.
$Id$
*/
/// @file iohandler/curl_io_handler.cc
#define GRB_LOG_FAC GrbLogFacility::curl
#ifdef HAVE_CURL
#include "curl_io_handler.h" // API
#include "config/config.h"
#include "exceptions.h"
#include <utility>
CurlIOHandler::CurlIOHandler(
const std::shared_ptr<Config>& config,
std::string url,
std::size_t bufSize,
std::size_t initialFillSize,
std::chrono::seconds connectTimeout,
std::chrono::seconds timeout)
: IOHandlerBufferHelper(config, bufSize, initialFillSize)
, URL(std::move(url))
, connectTimeout(connectTimeout)
, timeout(timeout)
{
if (this->URL.empty())
throw_std_runtime_error("URL has not been set correctly");
if (bufSize < CURL_MAX_WRITE_SIZE)
throw_std_runtime_error("bufSize must be at least CURL_MAX_WRITE_SIZE({})", CURL_MAX_WRITE_SIZE);
signalAfterEveryRead = true;
// still todo: optimize seek if data already in buffer
seekEnabled = true;
}
CurlIOHandler::~CurlIOHandler() noexcept
{
CurlIOHandler::close();
}
void CurlIOHandler::open(enum UpnpOpenFileMode mode)
{
if (!curl_handle) {
curl_handle = curl_easy_init();
if (!curl_handle)
throw_std_runtime_error("failed to init curl");
} else
curl_easy_reset(curl_handle);
IOHandlerBufferHelper::open(mode);
}
void CurlIOHandler::close()
{
if (isOpen)
IOHandlerBufferHelper::close();
if (!external_curl_handle && curl_handle)
curl_easy_cleanup(curl_handle);
}
void CurlIOHandler::threadProc()
{
StdThreadRunner::waitFor("CurlIOHandler", [this] { return threadRunner != nullptr; });
CURLcode res;
assert(curl_handle);
assert(!URL.empty());
curl_easy_setopt(curl_handle, CURLOPT_URL, URL.c_str());
curl_easy_setopt(curl_handle, CURLOPT_NOSIGNAL, 1);
curl_easy_setopt(curl_handle, CURLOPT_FOLLOWLOCATION, 1);
curl_easy_setopt(curl_handle, CURLOPT_MAXREDIRS, -1);
#ifdef GRBDEBUG
bool logEnabled = GrbLogger::Logger.isDebugging(GRB_LOG_FAC) || GrbLogger::Logger.isDebugLogging();
#else
#ifdef TOMBDEBUG
bool logEnabled = !GrbLogger::Logger.isDebugLogging();
#else
bool logEnabled = GrbLogger::Logger.isDebugLogging();
#endif
#endif
char errorBuffer[CURL_ERROR_SIZE] = { '\0' };
if (logEnabled) {
curl_easy_setopt(curl_handle, CURLOPT_ERRORBUFFER, errorBuffer);
curl_easy_setopt(curl_handle, CURLOPT_VERBOSE, 1);
}
curl_easy_setopt(curl_handle, CURLOPT_CONNECTTIMEOUT, connectTimeout.count()); // seconds
curl_easy_setopt(curl_handle, CURLOPT_TIMEOUT, timeout.count()); // seconds
// proxy..
curl_easy_setopt(curl_handle, CURLOPT_WRITEFUNCTION, CurlIOHandler::curlCallback);
curl_easy_setopt(curl_handle, CURLOPT_WRITEDATA, this);
auto lock = threadRunner->uniqueLock(std::defer_lock);
do {
lock.lock();
if (doSeek) {
log_debug("SEEK: {} {}", seekOffset, seekWhence);
if (seekWhence == SEEK_SET) {
posRead = seekOffset;
curl_easy_setopt(curl_handle, CURLOPT_RESUME_FROM_LARGE, seekOffset);
} else if (seekWhence == SEEK_CUR) {
posRead += seekOffset;
curl_easy_setopt(curl_handle, CURLOPT_RESUME_FROM_LARGE, posRead);
} else {
log_error("CurlIOHandler currently does not support SEEK_END");
static_assert(true);
}
/// \todo should we do that?
waitForInitialFillSize = (initialFillSize > 0);
doSeek = false;
threadRunner->notify();
}
lock.unlock();
res = curl_easy_perform(curl_handle);
} while (doSeek);
if (res != CURLE_OK) {
readError = true;
if (logEnabled) {
std::size_t len = strlen(errorBuffer);
if (len) {
log_error("libcurl (error {}): {}", res, errorBuffer);
} else
log_error("libcurl (error {}): {}", res, curl_easy_strerror(res));
}
} else
eof = true;
threadRunner->notify();
}
std::size_t CurlIOHandler::curlCallback(void* ptr, std::size_t size, std::size_t nmemb, CurlIOHandler* ego)
{
std::size_t wantWrite = size * nmemb;
assert(wantWrite <= ego->bufSize);
auto&& threadRunner = ego->threadRunner;
log_debug("URL: {}; size: {}; nmemb: {}; wantWrite: {}", ego->URL.c_str(), size, nmemb, wantWrite);
auto lock = threadRunner->uniqueLock();
bool first = true;
int bufFree;
do {
if (ego->doSeek && !ego->empty && (ego->seekWhence == SEEK_SET || (ego->seekWhence == SEEK_CUR && ego->seekOffset > 0))) {
auto currentFillSize = static_cast<int>(ego->endIndex - ego->startIndex);
if (currentFillSize <= 0)
currentFillSize += ego->bufSize;
auto relSeek = static_cast<int>(ego->seekOffset);
if (ego->seekWhence == SEEK_SET)
relSeek -= ego->posRead;
if (relSeek <= currentFillSize) { // we have everything we need in the buffer already
ego->startIndex += relSeek;
ego->posRead += relSeek;
if (ego->startIndex >= ego->bufSize)
ego->startIndex -= ego->bufSize;
if (ego->startIndex == ego->endIndex) {
ego->empty = true;
ego->startIndex = ego->endIndex = 0;
}
/// \todo do we need to wait for initialFillSize again?
ego->doSeek = false;
threadRunner->notify();
}
}
// note: seeking could be optimized some more (backward seeking)
// but this should suffice for now
if (ego->doSeek) { // seek not been processed yet
ego->startIndex = ego->endIndex = 0;
ego->empty = true;
// terminate this request, because we need a new request
// after the seek
return 0;
}
if (!first) {
threadRunner->wait(lock);
} else
first = false;
if (ego->threadShutdown)
return 0;
if (ego->empty) {
ego->startIndex = ego->endIndex = 0;
bufFree = ego->bufSize;
} else {
bufFree = ego->startIndex - ego->endIndex;
if (bufFree < 0)
bufFree += ego->bufSize;
}
} while (static_cast<std::size_t>(bufFree) < wantWrite);
std::size_t maxWrite = (ego->empty ? ego->bufSize : (ego->startIndex < ego->endIndex ? ego->bufSize - ego->endIndex : ego->startIndex - ego->endIndex));
std::size_t write1 = (wantWrite > maxWrite ? maxWrite : wantWrite);
std::size_t write2 = (write1 < wantWrite ? wantWrite - write1 : 0);
std::size_t bLocal = ego->endIndex;
lock.unlock();
std::memcpy(ego->buffer + bLocal, ptr, write1);
if (write2)
std::memcpy(ego->buffer, static_cast<char*>(ptr) + maxWrite, write2);
lock.lock();
#if 0
ego->bytesCurl += wantWrite;
#endif
ego->endIndex += wantWrite;
if (ego->endIndex >= ego->bufSize)
ego->endIndex -= ego->bufSize;
if (ego->empty) {
ego->empty = false;
threadRunner->notify();
}
if (ego->waitForInitialFillSize) {
auto currentFillSize = static_cast<int>(ego->endIndex - ego->startIndex);
if (currentFillSize <= 0)
currentFillSize += ego->bufSize;
if (static_cast<std::size_t>(currentFillSize) >= ego->initialFillSize) {
log_debug("buffer: initial fillsize reached");
ego->waitForInitialFillSize = false;
threadRunner->notify();
}
}
return wantWrite;
}
#endif // HAVE_CURL