From 00cc4ebe28b5d5660fa88a624506ab5f5cbff267 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Adrien=20B=C3=A9raud?= <adrien.beraud@savoirfairelinux.com>
Date: Sun, 17 Feb 2019 02:52:49 -0500
Subject: [PATCH] proxy server: send push notification on thread pool
---
include/opendht/dht_proxy_server.h | 2 ++
src/dht_proxy_server.cpp | 24 ++++++++++++++----------
2 files changed, 16 insertions(+), 10 deletions(-)
diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h
index a9552d8e..14b3d24f 100644
--- a/include/opendht/dht_proxy_server.h
+++ b/include/opendht/dht_proxy_server.h
@@ -45,6 +45,7 @@ namespace Json {
namespace dht {
class DhtRunner;
+class ThreadPool;
/**
* Describes the REST API
@@ -277,6 +278,7 @@ private:
std::condition_variable schedulerCv_;
Scheduler scheduler_;
std::thread schedulerThread_;
+ std::unique_ptr<ThreadPool> threadPool_;
Sp<Scheduler::Job> printStatsJob_;
mutable std::mutex statsMutex_;
diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp
index ab7b7326..2a241267 100644
--- a/src/dht_proxy_server.cpp
+++ b/src/dht_proxy_server.cpp
@@ -20,6 +20,7 @@
#if OPENDHT_PROXY_SERVER
#include "dht_proxy_server.h"
+#include "thread_pool.h"
#include "default_types.h"
#include "dhtrunner.h"
@@ -47,9 +48,11 @@ struct DhtProxyServer::SearchPuts {
};
constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2};
+constexpr const size_t IO_THREADS_MAX {64};
+
DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port , const std::string& pushServer)
-: dht_(dht) , pushServer_(pushServer)
+: dht_(dht), threadPool_(new ThreadPool(IO_THREADS_MAX)), pushServer_(pushServer)
{
if (not dht_)
throw std::invalid_argument("A DHT instance must be provided");
@@ -177,6 +180,7 @@ DhtProxyServer::stop()
schedulerThread_.join();
if (server_thread.joinable())
server_thread.join();
+ threadPool_->stop();
}
void
@@ -441,9 +445,7 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
Json::StreamWriterBuilder wbuilder;
wbuilder["commentStyle"] = "None";
wbuilder["indentation"] = "";
- auto output = Json::writeString(
- wbuilder, value->toJson()) +
- "\n";
+ auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
s->yield(output, [](const Sp<restbed::Session>
& /*session*/) {});
return true;
@@ -473,12 +475,14 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
// The listener is not found, so add it.
listener.internalToken = dht_->listen(infoHash,
- [this, infoHash, pushToken, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
- // Build message content.
- Json::Value json;
- json["key"] = infoHash.toString();
- json["to"] = clientId;
- sendPushNotification(pushToken, std::move(json), isAndroid);
+ [this, infoHash, pushToken, isAndroid, clientId](const std::vector<std::shared_ptr<Value>>& /*values*/) {
+ threadPool_->run([this, infoHash, pushToken, isAndroid, clientId](){
+ // Build message content
+ Json::Value json;
+ json["key"] = infoHash.toString();
+ json["to"] = clientId;
+ sendPushNotification(pushToken, std::move(json), isAndroid);
+ });
return true;
}
);
--
GitLab