Reworked http client to ease reuse

This commit is contained in:
emeric
2021-10-05 18:41:54 +02:00
parent 617139c2f4
commit 11f8a9e38b
16 changed files with 438 additions and 238 deletions
-1
View File
@@ -3,7 +3,6 @@ add_library(lmsscrobbling SHARED
impl/internal/InternalScrobbler.cpp
impl/listenbrainz/ListenBrainzScrobbler.cpp
impl/listenbrainz/ListensSynchronizer.cpp
impl/listenbrainz/SendQueue.cpp
impl/listenbrainz/Utils.cpp
impl/Scrobbling.cpp
)
@@ -32,6 +32,7 @@
#include "database/TrackList.hpp"
#include "database/User.hpp"
#include "utils/IConfig.hpp"
#include "utils/http/IClient.hpp"
#include "utils/Logger.hpp"
#include "utils/Service.hpp"
#include "Utils.hpp"
@@ -142,10 +143,10 @@ namespace Scrobbling::ListenBrainz
Scrobbler::Scrobbler(boost::asio::io_context& ioContext, Database::Db& db)
: _ioContext {ioContext}
, _db {db}
, _sendQueue {_ioContext, Service<IConfig>::get()->getString("listenbrainz-api-base-url", "https://api.listenbrainz.org")}
, _listensSynchronizer {_ioContext, db, _sendQueue}
, _baseAPIUrl {Service<IConfig>::get()->getString("listenbrainz-api-base-url", "https://api.listenbrainz.org")}
, _listensSynchronizer {_ioContext, db, _baseAPIUrl}
{
LOG(INFO) << "Starting ListenBrainz scrobbler... API endpoint = '" << _sendQueue.getAPIBaseURL();
LOG(INFO) << "Starting ListenBrainz scrobbler... API endpoint = '" << _baseAPIUrl;
}
Scrobbler::~Scrobbler()
@@ -187,53 +188,38 @@ namespace Scrobbling::ListenBrainz
void
Scrobbler::enqueListen(const Listen& listen, const Wt::WDateTime& timePoint)
{
std::optional<SendQueue::RequestData> requestData {createSubmitListenRequestData(listen, timePoint)};
if (!requestData)
return;
Http::ClientPOSTRequestParameters request;
request.url = _baseAPIUrl + "/1/submit-listens";
SendQueue::Request submitListen {std::move(*requestData)};
if (timePoint.isValid())
{
submitListen.setPriority(SendQueue::Request::Priority::Normal);
submitListen.setOnSuccessFunc([=](std::string_view)
request.priority = Http::ClientRequestParameters::Priority::Normal;
request.onSuccessFunc = [=](std::string_view)
{
_listensSynchronizer.saveListen(TimedListen {listen, timePoint});
});
};
}
else
{
// We want "listen now" to appear as soon as possible
submitListen.setPriority(SendQueue::Request::Priority::High);
request.priority = Http::ClientRequestParameters::Priority::High;
}
_sendQueue.enqueueRequest(std::move(submitListen));
}
std::optional<SendQueue::RequestData>
Scrobbler::createSubmitListenRequestData(const Listen& listen, const Wt::WDateTime& timePoint)
{
Database::Session& session {_db.getTLSSession()};
const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(session, listen.userId)};
if (!listenBrainzToken)
return std::nullopt;
SendQueue::RequestData requestData;
requestData.endpoint = "/1/submit-listens";
requestData.type = SendQueue::RequestData::Type::POST;
std::string bodyText {listenToJsonString(session, listen, timePoint, timePoint.isValid() ? "single" : "playing_now")};
std::string bodyText {listenToJsonString(_db.getTLSSession(), listen, timePoint, timePoint.isValid() ? "single" : "playing_now")};
if (bodyText.empty())
{
LOG(DEBUG) << "Cannot convert listen to json: skipping";
return std::nullopt;
return;
}
requestData.message.addBodyText(bodyText);
requestData.message.addHeader("Authorization", "Token " + std::string {listenBrainzToken->getAsString()});
requestData.message.addHeader("Content-Type", "application/json");
const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(_db.getTLSSession(), listen.userId)};
if (!listenBrainzToken)
return;
return requestData;
request.message.addBodyText(bodyText);
request.message.addHeader("Authorization", "Token " + std::string {listenBrainzToken->getAsString()});
request.message.addHeader("Content-Type", "application/json");
Service<Http::IClient>::get()->sendPOSTRequest(std::move(request));
}
} // namespace Scrobbling::ListenBrainz
@@ -24,7 +24,6 @@
#include "IScrobbler.hpp"
#include "ListensSynchronizer.hpp"
#include "SendQueue.hpp"
namespace Database
{
@@ -54,11 +53,11 @@ namespace Scrobbling::ListenBrainz
// Submit listens
void enqueListen(const Listen& listen, const Wt::WDateTime& timePoint);
std::optional<SendQueue::RequestData> createSubmitListenRequestData(const Listen& listen, const Wt::WDateTime& timePoint);
//std::optional<SendQueue::RequestData> createSubmitListenRequestData(const Listen& listen, const Wt::WDateTime& timePoint);
boost::asio::io_context& _ioContext;
Database::Db& _db;
SendQueue _sendQueue;
std::string _baseAPIUrl;
ListensSynchronizer _listensSynchronizer;
};
} // Scrobbling::ListenBrainz
@@ -34,6 +34,7 @@
#include "database/User.hpp"
#include "scrobbling/Exception.hpp"
#include "utils/IConfig.hpp"
#include "utils/http/IClient.hpp"
#include "utils/Logger.hpp"
#include "utils/Service.hpp"
@@ -45,17 +46,6 @@ namespace
{
using namespace Scrobbling::ListenBrainz;
SendQueue::RequestData
createValidateTokenRequestData(std::string_view authToken)
{
SendQueue::RequestData requestData;
requestData.type = SendQueue::RequestData::Type::GET;
requestData.endpoint = "/1/validate-token";
requestData.headers = { {"Authorization", "Token " + std::string {authToken}} };
return requestData;
}
std::string
parseValidateToken(std::string_view msgBody)
{
@@ -79,18 +69,6 @@ namespace
return listenBrainzUserName;
}
SendQueue::RequestData
createListenCountRequestData(std::string_view listenBrainzUserName)
{
LOG(DEBUG) << "Getting listen count for listenbrainz user '" << listenBrainzUserName << "'";
SendQueue::RequestData requestData;
requestData.type = SendQueue::RequestData::Type::GET;
requestData.endpoint = "/1/user/" + std::string {listenBrainzUserName} + "/listen-count";
return requestData;
}
std::optional<std::size_t>
parseListenCount(std::string_view msgBody)
{
@@ -109,18 +87,6 @@ namespace
}
}
SendQueue::RequestData
createGetListensRequestData(std::string_view listenBrainzUserName, const Wt::WDateTime& maxDateTime)
{
LOG(DEBUG) << "Getting listens for listenbrainz user '" << listenBrainzUserName << "' with max_ts = " << maxDateTime.toString();
SendQueue::RequestData requestData;
requestData.type = SendQueue::RequestData::Type::GET;
requestData.endpoint = "/1/user/" + std::string {listenBrainzUserName} + "/listens?max_ts=" + std::to_string(maxDateTime.toTime_t());
return requestData;
}
Database::Track::pointer
tryMatchListen(Database::Session& session, const Wt::Json::Object& metadata)
{
@@ -247,10 +213,10 @@ namespace
namespace Scrobbling::ListenBrainz
{
ListensSynchronizer::ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, SendQueue& sendQueue)
ListensSynchronizer::ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, std::string_view baseAPIUrl)
: _ioContext {ioContext}
, _db {db}
, _sendQueue {sendQueue}
, _baseAPIUrl {baseAPIUrl}
, _maxSyncListenCount {Service<IConfig>::get()->getULong("listenbrainz-max-sync-listen-count", 1000)}
, _syncListensPeriod {Service<IConfig>::get()->getULong("listenbrainz-sync-listens-period-hours", 1)}
{
@@ -386,31 +352,33 @@ namespace Scrobbling::ListenBrainz
{
assert(context.listenBrainzUserName.empty());
std::optional<SendQueue::RequestData> requestData {createValidateTokenRequestData(context.userId)};
if (!requestData)
const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(_db.getTLSSession(), context.userId)};
if (!listenBrainzToken)
{
onGetListensEnded(context);
return;
}
SendQueue::Request validateTokenRequest {std::move(*requestData)};
validateTokenRequest.setOnSuccessFunc([this, &context] (std::string_view msgBody)
{
context.listenBrainzUserName = parseValidateToken(msgBody);
if (context.listenBrainzUserName.empty())
Http::ClientGETRequestParameters request;
request.priority = Http::ClientRequestParameters::Priority::Low;
request.url = _baseAPIUrl + "/1/validate-token";
request.headers = { {"Authorization", "Token " + std::string {listenBrainzToken->getAsString()}} };
request.onSuccessFunc = [this, &context] (std::string_view msgBody)
{
context.listenBrainzUserName = parseValidateToken(msgBody);
if (context.listenBrainzUserName.empty())
{
onGetListensEnded(context);
return;
}
enqueGetListenCount(context);
};
request.onFailureFunc = [this, &context]
{
onGetListensEnded(context);
return;
}
enqueGetListenCount(context);
});
validateTokenRequest.setOnFailureFunc([this, &context]
{
onGetListensEnded(context);
});
};
validateTokenRequest.setPriority(SendQueue::Request::Priority::Low);
_sendQueue.enqueueRequest(std::move(validateTokenRequest));
Service<Http::IClient>::get()->sendGETRequest(std::move(request));
}
void
@@ -418,32 +386,33 @@ namespace Scrobbling::ListenBrainz
{
assert(!context.listenBrainzUserName.empty());
SendQueue::Request getListenCountRequest {createListenCountRequestData(context.listenBrainzUserName)};
getListenCountRequest.setOnSuccessFunc([=, &context] (std::string_view msgBody)
{
const auto listenCount = parseListenCount(msgBody);
if (listenCount)
LOG(DEBUG) << "Listen count for listenbrainz user '" << context.listenBrainzUserName << "' = " << *listenCount;
Http::ClientGETRequestParameters request;
request.url = _baseAPIUrl + "/1/user/" + std::string {context.listenBrainzUserName} + "/listen-count";
request.priority = Http::ClientRequestParameters::Priority::Low;
request.onSuccessFunc = [=, &context] (std::string_view msgBody)
{
const auto listenCount = parseListenCount(msgBody);
if (listenCount)
LOG(DEBUG) << "Listen count for listenbrainz user '" << context.listenBrainzUserName << "' = " << *listenCount;
bool needSync {listenCount && (!context.listenCount || *context.listenCount != *listenCount)};
context.listenCount = listenCount;
bool needSync {listenCount && (!context.listenCount || *context.listenCount != *listenCount)};
context.listenCount = listenCount;
if (!needSync)
if (!needSync)
{
onGetListensEnded(context);
return;
}
context.maxDateTime = Wt::WDateTime::currentDateTime();
enqueGetListens(context);
};
request.onFailureFunc = [this, &context]
{
onGetListensEnded(context);
return;
}
};
context.maxDateTime = Wt::WDateTime::currentDateTime();
enqueGetListens(context);
});
getListenCountRequest.setOnFailureFunc([this, &context]
{
onGetListensEnded(context);
});
getListenCountRequest.setPriority(SendQueue::Request::Priority::Low);
_sendQueue.enqueueRequest(std::move(getListenCountRequest));
Service<Http::IClient>::get()->sendGETRequest(std::move(request));
}
void
@@ -451,37 +420,26 @@ namespace Scrobbling::ListenBrainz
{
assert(!context.listenBrainzUserName.empty());
SendQueue::Request getListensRequest {::createGetListensRequestData(context.listenBrainzUserName, context.maxDateTime)};
getListensRequest.setOnSuccessFunc([=, &context] (std::string_view msgBody)
{
processGetListensResponse(msgBody, context);
if (context.fetchedListenCount >= _maxSyncListenCount || !context.maxDateTime.isValid())
Http::ClientGETRequestParameters request;
request.url = _baseAPIUrl + "/1/user/" + context.listenBrainzUserName + "/listens?max_ts=" + std::to_string(context.maxDateTime.toTime_t());
request.priority = Http::ClientRequestParameters::Priority::Low;
request.onSuccessFunc = [=, &context] (std::string_view msgBody)
{
processGetListensResponse(msgBody, context);
if (context.fetchedListenCount >= _maxSyncListenCount || !context.maxDateTime.isValid())
{
onGetListensEnded(context);
return;
}
enqueGetListens(context);
};
request.onFailureFunc = [=, &context]
{
onGetListensEnded(context);
return;
}
};
enqueGetListens(context);
});
getListensRequest.setOnFailureFunc([=, &context]
{
onGetListensEnded(context);
});
getListensRequest.setPriority(SendQueue::Request::Priority::Low);
_sendQueue.enqueueRequest(std::move(getListensRequest));
}
std::optional<SendQueue::RequestData>
ListensSynchronizer::createValidateTokenRequestData(Database::UserId userId)
{
Database::Session& session {_db.getTLSSession()};
const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(session, userId)};
if (!listenBrainzToken)
return std::nullopt;
return ::createValidateTokenRequestData(listenBrainzToken->getAsString());
Service<Http::IClient>::get()->sendGETRequest(std::move(request));
}
void
@@ -28,7 +28,6 @@
#include "database/Types.hpp"
#include "scrobbling/Listen.hpp"
#include "SendQueue.hpp"
namespace Database
{
@@ -43,7 +42,7 @@ namespace Scrobbling::ListenBrainz
class ListensSynchronizer
{
public:
ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, SendQueue& sendQueue);
ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, std::string_view baseAPIUrl);
void saveListen(const TimedListen& listen);
@@ -78,14 +77,12 @@ namespace Scrobbling::ListenBrainz
void enqueValidateToken(UserContext& context);
void enqueGetListenCount(UserContext& context);
void enqueGetListens(UserContext& context);
std::optional<SendQueue::RequestData> createValidateTokenRequestData(Database::UserId userId);
std::optional<SendQueue::RequestData> createGetListensRequestData(std::string_view listenBrainzUserName, const Wt::WDateTime& maxDateTime);
void processGetListensResponse(std::string_view body, UserContext& context);
boost::asio::io_context& _ioContext;
boost::asio::io_context::strand _strand {_ioContext};
Database::Db& _db;
SendQueue& _sendQueue;
std::string _baseAPIUrl;
boost::asio::steady_timer _getListensTimer {_ioContext};
std::unordered_map<Database::UserId, UserContext> _userContexts;
+2
View File
@@ -1,4 +1,6 @@
add_library(lmsutils SHARED
impl/http/Client.cpp
impl/http/SendQueue.cpp
impl/ChildProcess.cpp
impl/ChildProcessManager.cpp
impl/Config.cpp
+1
View File
@@ -31,6 +31,7 @@ const char* getModuleName(Module mod)
case Module::DB: return "DB";
case Module::DBUPDATER: return "DB UPDATER";
case Module::FEATURE: return "FEATURE";
case Module::HTTP: return "HTTP";
case Module::MAIN: return "MAIN";
case Module::METADATA: return "METADATA";
case Module::REMOTE: return "REMOTE";
+73
View File
@@ -0,0 +1,73 @@
/*
* Copyright (C) 2021 Emeric Poupon
*
* This file is part of LMS.
*
* LMS 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 3 of the License, or
* (at your option) any later version.
*
* LMS 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 LMS. If not, see <http://www.gnu.org/licenses/>.
*/
#include "Client.hpp"
#include "utils/Exception.hpp"
namespace Http
{
std::unique_ptr<IClient>
createClient(boost::asio::io_context& ioContext)
{
return std::make_unique<Client>(ioContext);
}
void
Client::sendGETRequest(ClientGETRequestParameters&& GETParams)
{
SendQueue& sendQueue {getOrCreateSendQueue(GETParams.url)};
sendQueue.sendRequest(std::make_unique<ClientRequest>(std::move(GETParams)));
}
void
Client::sendPOSTRequest(ClientPOSTRequestParameters&& POSTParams)
{
SendQueue& sendQueue {getOrCreateSendQueue(POSTParams.url)};
sendQueue.sendRequest(std::make_unique<ClientRequest>(std::move(POSTParams)));
}
SendQueue&
Client::getOrCreateSendQueue(const std::string& url)
{
Wt::Http::Client::URL parsedURL;
if (!Wt::Http::Client::parseUrl(url, parsedURL))
throw LmsException {"Cannot parse URL '" + url + "'"};
{
std::shared_lock lock {_sendQueuesMutex};
if (auto it = _sendQueues.find(parsedURL.host); it != std::cend(_sendQueues))
return it->second;
}
{
std::unique_lock lock {_sendQueuesMutex};
if (auto it = _sendQueues.find(parsedURL.host); it != std::cend(_sendQueues))
return it->second;
auto [it, inserted] {_sendQueues.emplace(parsedURL.host, _ioContext)};
assert(inserted);
return it->second;
}
}
} // namespace Http
+47
View File
@@ -0,0 +1,47 @@
/*
* Copyright (C) 2021 Emeric Poupon
*
* This file is part of LMS.
*
* LMS 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 3 of the License, or
* (at your option) any later version.
*
* LMS 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 LMS. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include <unordered_map>
#include <string>
#include <shared_mutex>
#include "utils/http/IClient.hpp"
#include "SendQueue.hpp"
namespace Http
{
class Client final : public IClient
{
public:
Client(boost::asio::io_context& ioContext) : _ioContext {ioContext} {}
private:
void sendGETRequest(ClientGETRequestParameters&& request) override;
void sendPOSTRequest(ClientPOSTRequestParameters&& request) override;
SendQueue& getOrCreateSendQueue(const std::string& host);
boost::asio::io_context& _ioContext;
std::shared_mutex _sendQueuesMutex;
std::unordered_map<std::string, SendQueue> _sendQueues;
};
} // namespace Http
@@ -0,0 +1,75 @@
/*
* Copyright (C) 2021 Emeric Poupon
*
* This file is part of LMS.
*
* LMS 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 3 of the License, or
* (at your option) any later version.
*
* LMS 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 LMS. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include <memory>
#include <variant>
#include "utils/http/ClientRequestParameters.hpp"
namespace Http
{
class ClientRequest
{
public:
ClientRequest(ClientGETRequestParameters&& GETParams) : _parameters {std::move(GETParams)} {}
ClientRequest(ClientPOSTRequestParameters&& POSTParams) : _parameters {std::move(POSTParams)} {}
std::size_t retryCount {};
const ClientRequestParameters& getParameters() const
{
const ClientRequestParameters* res;
std::visit([&](const auto& parameters)
{
res = &static_cast<const ClientRequestParameters&>(parameters);
}, _parameters);
return *res;
}
enum class Type
{
GET,
POST
};
Type getType() const
{
if (std::holds_alternative<ClientGETRequestParameters>(_parameters))
return Type::GET;
else
return Type::POST;
}
const ClientGETRequestParameters& getGETParameters() const
{
return std::get<ClientGETRequestParameters>(_parameters);
}
const ClientPOSTRequestParameters& getPOSTParameters() const
{
return std::get<ClientPOSTRequestParameters>(_parameters);
}
private:
std::variant<ClientGETRequestParameters, ClientPOSTRequestParameters> _parameters;
};
}
@@ -19,9 +19,10 @@
#include "SendQueue.hpp"
#include <boost/asio/dispatch.hpp>
#include <boost/asio/bind_executor.hpp>
#include "scrobbling/Exception.hpp"
#include "utils/Exception.hpp"
#include "utils/Logger.hpp"
#include "utils/String.hpp"
@@ -57,11 +58,10 @@ namespace
}
}
namespace Scrobbling::ListenBrainz
namespace Http
{
SendQueue::SendQueue(boost::asio::io_context& ioContext, std::string_view apiBaseURL)
SendQueue::SendQueue(boost::asio::io_context& ioContext)
: _ioContext {ioContext}
, _apiBaseURL {apiBaseURL}
{
_client.done().connect([this](Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg)
{
@@ -78,11 +78,11 @@ namespace Scrobbling::ListenBrainz
}
void
SendQueue::enqueueRequest(Request request)
SendQueue::sendRequest(std::unique_ptr<ClientRequest> request)
{
_strand.dispatch([this, request = std::move(request)]()
boost::asio::dispatch(_strand, [this, request = std::move(request)]() mutable
{
_sendQueue[request._priority].emplace_back(std::move(request));
_sendQueue[request->getParameters().priority].emplace_back(std::move(request));
if (_state == State::Idle)
sendNextQueuedRequest();
@@ -93,16 +93,17 @@ namespace Scrobbling::ListenBrainz
SendQueue::sendNextQueuedRequest()
{
assert(_state == State::Idle);
assert(!_currentRequest);
for (auto& [prio, requests] : _sendQueue)
{
LOG(DEBUG) << "Processing prio " << static_cast<int>(prio) << ", request count = " << requests.size();
while (!requests.empty())
{
Request request {std::move(requests.front())};
std::unique_ptr<ClientRequest> request {std::move(requests.front())};
requests.pop_front();
if (!sendRequest(request._requestData))
if (!sendRequest(*request))
continue;
_state = State::Sending;
@@ -113,20 +114,19 @@ namespace Scrobbling::ListenBrainz
}
bool
SendQueue::sendRequest(const RequestData& requestData)
SendQueue::sendRequest(const ClientRequest& request)
{
const std::string url {_apiBaseURL + requestData.endpoint};
LOG(DEBUG) << "Sending request to url '" << request.getParameters().url << "'";
LOG(DEBUG) << "Sending request type " << (requestData.type == RequestData::Type::GET ? "GET" : "POST") << " to url '" << url << "'";
bool res{};
switch (requestData.type)
bool res {};
switch (request.getType())
{
case RequestData::Type::GET:
res = _client.get(url, requestData.headers);
case ClientRequest::Type::GET:
res = _client.get(request.getParameters().url, request.getGETParameters().headers);
break;
case RequestData::Type::POST:
res = _client.post(url, requestData.message);
case ClientRequest::Type::POST:
res = _client.post(request.getParameters().url, request.getPOSTParameters().message);
break;
}
@@ -146,43 +146,43 @@ namespace Scrobbling::ListenBrainz
}
assert(_currentRequest);
Request request {std::move(*_currentRequest)};
_state = State::Idle;
LOG(DEBUG) << "Client done. status = " << msg.status();
if (ec)
onClientDoneError(std::move(request), ec);
onClientDoneError(std::move(_currentRequest), ec);
else
onClientDoneSuccess(std::move(request), msg);
onClientDoneSuccess(std::move(_currentRequest), msg);
}
void
SendQueue::onClientDoneError(Request request, Wt::AsioWrapper::error_code ec)
SendQueue::onClientDoneError(std::unique_ptr<ClientRequest> request, Wt::AsioWrapper::error_code ec)
{
LOG(ERROR) << "Retry " << request._retryCount << ", client error: '" << ec.message() << "'";
LOG(ERROR) << "Retry " << request->retryCount << ", client error: '" << ec.message() << "'";
// may be a network error, try again later
throttle(_defaultRetryWaitDuration);
if (request._retryCount++ < _maxRetryCount)
if (request->retryCount++ < _maxRetryCount)
{
_sendQueue[request._priority].emplace_front(std::move(request));
_sendQueue[request->getParameters().priority].emplace_front(std::move(request));
}
else
{
LOG(ERROR) << "Too many retries, giving up operation and throttle";
if (request._onFailureFunc)
request._onFailureFunc();
if (request->getParameters().onFailureFunc)
request->getParameters().onFailureFunc();
}
}
void
SendQueue::onClientDoneSuccess(Request request, const Wt::Http::Message& msg)
SendQueue::onClientDoneSuccess(std::unique_ptr<ClientRequest> request, const Wt::Http::Message& msg)
{
const ClientRequestParameters& requestParameters {request->getParameters()};
bool mustThrottle{};
if (msg.status() == 429)
{
_sendQueue[request._priority].emplace_front(std::move(request));
_sendQueue[requestParameters.priority].emplace_front(std::move(request));
mustThrottle = true;
}
@@ -198,14 +198,14 @@ namespace Scrobbling::ListenBrainz
{
if (msg.status() == 200)
{
if (request._onSuccessFunc)
request._onSuccessFunc(msg.body());
if (requestParameters.onSuccessFunc)
requestParameters.onSuccessFunc(msg.body());
}
else
{
LOG(ERROR) << "Send error: '" << msg.body() << "'";
if (request._onFailureFunc)
request._onFailureFunc();
if (requestParameters.onFailureFunc)
requestParameters.onFailureFunc();
}
}
@@ -231,7 +231,7 @@ namespace Scrobbling::ListenBrainz
}
else if (ec)
{
throw Exception {"Throttle timer failure: " + std::string {ec.message()} };
throw LmsException {"Throttle timer failure: " + std::string {ec.message()} };
}
_state = State::Idle;
@@ -20,19 +20,22 @@
#pragma once
#include <deque>
#include <vector>
#include <string_view>
#include <boost/asio/io_context.hpp>
#include <boost/asio/io_context_strand.hpp>
#include <boost/asio/steady_timer.hpp>
#include <Wt/Http/Client.h>
#include "ClientRequest.hpp"
namespace Scrobbling::ListenBrainz
namespace Http
{
class SendQueue
{
public:
SendQueue(boost::asio::io_context& ioContext, std::string_view apiBaseURL);
SendQueue(boost::asio::io_context& ioContext);
~SendQueue();
SendQueue(const SendQueue&) = delete;
@@ -40,58 +43,15 @@ namespace Scrobbling::ListenBrainz
SendQueue& operator=(const SendQueue&) = delete;
SendQueue& operator=(const SendQueue&&) = delete;
// generic queue operations
struct RequestData
{
enum class Type
{
GET,
POST,
};
Type type;
std::string endpoint; // relative URL to the base API
std::vector<Wt::Http::Message::Header> headers; // used by GET
Wt::Http::Message message; // used by POST
};
class Request
{
public:
enum class Priority
{
High,
Normal,
Low,
};
Request(RequestData requestData) : _requestData {std::move(requestData)} {}
using OnSuccessFunc = std::function<void(std::string_view msgBody)>;
using OnFailureFunc = std::function<void()>;
void setOnSuccessFunc(OnSuccessFunc onSuccessFunc) { _onSuccessFunc = onSuccessFunc; }
void setOnFailureFunc(OnFailureFunc onFailureFunc) { _onFailureFunc = onFailureFunc; }
void setPriority(Priority priority) { _priority = priority; }
private:
friend class SendQueue;
RequestData _requestData;
Priority _priority {Priority::Normal};
std::size_t _retryCount {};
OnSuccessFunc _onSuccessFunc;
OnFailureFunc _onFailureFunc;
};
std::string_view getAPIBaseURL() const { return _apiBaseURL; }
void enqueueRequest(Request request);
void sendRequest(std::unique_ptr<ClientRequest> request);
private:
void sendNextQueuedRequest();
bool sendRequest(const RequestData& request);
bool sendRequest(const ClientRequest& request);
void onClientDone(Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg);
void onClientDoneError(Request request, Wt::AsioWrapper::error_code ec);
void onClientDoneSuccess(Request request, const Wt::Http::Message& msg);
void onClientDoneError(std::unique_ptr<ClientRequest> request, Wt::AsioWrapper::error_code ec);
void onClientDoneSuccess(std::unique_ptr<ClientRequest> request, const Wt::Http::Message& msg);
void throttle(std::chrono::seconds duration);
const std::size_t _maxRetryCount {2};
@@ -99,20 +59,20 @@ namespace Scrobbling::ListenBrainz
const std::chrono::seconds _minRetryWaitDuration {1};
const std::chrono::seconds _maxRetryWaitDuration {300};
boost::asio::io_context& _ioContext;
boost::asio::io_context::strand _strand {_ioContext};
boost::asio::steady_timer _throttleTimer {_ioContext};
enum class State
{
Idle,
Throttled,
Sending,
};
boost::asio::io_context& _ioContext;
boost::asio::io_context::strand _strand {_ioContext};
boost::asio::steady_timer _throttleTimer {_ioContext};
std::string _apiBaseURL;
State _state {State::Idle};
Wt::Http::Client _client {_ioContext};
std::map<Request::Priority, std::deque<Request>> _sendQueue;
std::optional<Request> _currentRequest;
State _state {State::Idle};
Wt::Http::Client _client {_ioContext};
std::map<ClientRequestParameters::Priority, std::deque<std::unique_ptr<ClientRequest>>> _sendQueue;
std::unique_ptr<ClientRequest> _currentRequest;
};
} // namespace Scrobbling::ListenBrainz
+1
View File
@@ -43,6 +43,7 @@ enum class Module
DB,
DBUPDATER,
FEATURE,
HTTP,
MAIN,
METADATA,
REMOTE,
@@ -0,0 +1,60 @@
/*
* Copyright (C) 2021 Emeric Poupon
*
* This file is part of LMS.
*
* LMS 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 3 of the License, or
* (at your option) any later version.
*
* LMS 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 LMS. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include <functional>
#include <string_view>
#include <vector>
#include <Wt/Http/Message.h>
namespace Http
{
struct ClientRequestParameters
{
enum class Priority
{
High,
Normal,
Low,
};
Priority priority {Priority::Normal};
std::string url;
using OnSuccessFunc = std::function<void(std::string_view msgBody)>;
OnSuccessFunc onSuccessFunc;
using OnFailureFunc = std::function<void()>;
OnFailureFunc onFailureFunc;
};
struct ClientGETRequestParameters final : public ClientRequestParameters
{
std::vector<Wt::Http::Message::Header> headers;
};
struct ClientPOSTRequestParameters final : public ClientRequestParameters
{
Wt::Http::Message message;
};
} // namespace Http
@@ -0,0 +1,40 @@
/*
* Copyright (C) 2021 Emeric Poupon
*
* This file is part of LMS.
*
* LMS 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 3 of the License, or
* (at your option) any later version.
*
* LMS 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 LMS. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include <string_view>
#include <boost/asio/io_context.hpp>
#include "utils/http/ClientRequestParameters.hpp"
namespace Http
{
class IClient
{
public:
virtual ~IClient() = default;
virtual void sendGETRequest(ClientGETRequestParameters&& request) = 0;
virtual void sendPOSTRequest(ClientPOSTRequestParameters&& request) = 0;
};
std::unique_ptr<IClient> createClient(boost::asio::io_context& ioContext);
} // namespace Http
+2
View File
@@ -37,6 +37,7 @@
#include "scrobbling/IScrobbling.hpp"
#include "ui/LmsApplication.hpp"
#include "ui/LmsApplicationManager.hpp"
#include "utils/http/IClient.hpp"
#include "utils/IChildProcessManager.hpp"
#include "utils/IConfig.hpp"
#include "utils/IOContextRunner.hpp"
@@ -255,6 +256,7 @@ int main(int argc, char* argv[])
else
throw LmsException {"Bad value '" + authenticationBackend + "' for 'authentication-backend'"};
Service<Http::IClient> httpClient {Http::createClient(ioContext)};
Service<CoverArt::IGrabber> coverArtService {CoverArt::createGrabber(database, argv[0], server.appRoot() + "/images/unknown-cover.jpg")};
Service<Recommendation::IEngine> recommendationEngineService {Recommendation::createEngine(database)};
Service<Scanner::IScanner> scannerService {Scanner::createScanner(database, *recommendationEngineService)};