Compare commits

..
3 Commits
Author SHA1 Message Date
Adrien Béraud cf378c1a0e update version to 3.2.0 2024-05-25 23:59:20 -04:00
Adrien Béraud 9a987a0586 ci: build proxy client for Autotools build 2024-05-25 23:59:20 -04:00
Adrien Béraud 8806960d2b http: migrate from http-parser to llhttp 2024-05-25 23:59:20 -04:00
36 changed files with 176 additions and 643 deletions

No files matched your search

+6 -9
View File
@@ -7,10 +7,9 @@ jobs:
name: Ubuntu/GCC Autotools build
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: Dependencies
run: |
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev libasio-dev \
libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev
@@ -50,10 +49,9 @@ jobs:
name: Ubuntu/GCC Meson build
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: Dependencies
run: |
sudo apt update && \
sudo apt install ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev \
libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip \
libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-0-dev
@@ -68,10 +66,9 @@ jobs:
name: Ubuntu/GCC minimal build
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: Dependencies
run: |
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev libfmt-dev \
libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-0-dev
- name: Configure
@@ -87,16 +84,16 @@ jobs:
build-macos:
name: macOS/Clang build
runs-on: macOS-15
runs-on: macos-12
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: Dependencies
run: |
brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl cppunit
- name: expected-lite
run: |
sudo mkdir -p /usr/local/include/nonstd && \
sudo mkdir /usr/local/include/nonstd && \
sudo wget https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp \
-O /usr/local/include/nonstd/expected.hpp
+1 -2
View File
@@ -6,10 +6,9 @@ jobs:
name: Clang Static Analysis
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: deps
run: |
sudo apt update && \
sudo apt install libncurses5-dev libreadline-dev nettle-dev \
libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libfmt-dev libmsgpack-dev libargon2-0-dev libasio-dev \
+1 -1
View File
@@ -26,7 +26,7 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
# Initializes the CodeQL tools for scanning.
- name: Initialize CodeQL
+31 -31
View File
@@ -25,10 +25,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -36,12 +36,12 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DEPS }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/DockerfileDeps
@@ -58,10 +58,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -69,13 +69,13 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/Dockerfile
@@ -92,10 +92,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -103,12 +103,12 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DEPS_LLVM }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/DockerfileDepsLlvm
@@ -125,10 +125,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -136,13 +136,13 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME_LLVM }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/DockerfileLlvm
@@ -159,10 +159,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -170,13 +170,13 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DHTNODE }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/DockerfileDhtnode
@@ -193,10 +193,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -204,13 +204,13 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME_ALPINE_DEPS }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/DockerfileDepsAlpine
@@ -227,10 +227,10 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: Log in to the Container registry
uses: docker/login-action@v3
uses: docker/login-action@v2
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
@@ -238,13 +238,13 @@ jobs:
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
uses: docker/metadata-action@v4
with:
images: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME_ALPINE }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
uses: docker/build-push-action@v3
with:
context: .
file: docker/DockerfileAlpine
@@ -262,7 +262,7 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v3
- name: cmake
run: |
@@ -280,14 +280,14 @@ jobs:
- name: build
run: cd build && make dist
- uses: actions/upload-artifact@v4
- uses: actions/upload-artifact@v3
with:
name: opendht-wheels-linux
path: build/python/dist/*.whl
- name: Create Release
id: create_release
uses: softprops/action-gh-release@v2
uses: softprops/action-gh-release@v1
with:
draft: true
files: build/python/dist/*.whl
+1 -12
View File
@@ -85,9 +85,6 @@ if (NOT MSVC)
if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
message(STATUS "Found ASIO ${ASIO_INCLUDE_DIR}")
else()
message(STATUS "ASIO not required")
endif ()
find_package(fmt)
@@ -97,7 +94,7 @@ if (NOT MSVC)
# llhttp
find_path(LLHTTP_INCLUDE_DIR llhttp.h)
find_library(LLHTTP_LIBRARY NAMES llhttp libllhttp)
find_library(LLHTTP_LIBRARY libllhttp.a)
if (LLHTTP_INCLUDE_DIR AND LLHTTP_LIBRARY)
message(STATUS "Found llhttp ${LLHTTP_INCLUDE_DIR} ${LLHTTP_LIBRARY}")
add_library(llhttp_static STATIC IMPORTED)
@@ -163,9 +160,6 @@ endif ()
if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
add_definitions(-DASIO_STANDALONE)
if (UNIX AND NOT APPLE)
pkg_search_module(liburing IMPORTED_TARGET liburing)
endif ()
endif()
if (NOT MSVC)
@@ -407,11 +401,6 @@ else()
if (OPENDHT_PROXY_OPENSSL)
target_link_libraries(opendht PUBLIC PkgConfig::OPENSSL)
endif()
if (liburing_FOUND)
set(iouring_lib ", liburing")
target_link_libraries(opendht PUBLIC PkgConfig::liburing)
target_compile_definitions(opendht PUBLIC ASIO_HAS_IO_URING ASIO_DISABLE_EPOLL)
endif()
endif()
if (BUILD_SHARED_LIBS)
-1
View File
@@ -98,7 +98,6 @@ for r in results:
- Nettle 2.4+, a GnuTLS dependency for crypto.
- {fmt} 9.0+, for log formatting.
- (optional) restinio used for the REST API.
- (optional) llhttp used for the REST API.
- (optional) jsoncpp 1.7.4-3+, used for the REST API.
- Build tested with GCC 7+ (GNU/Linux, Windows with MinGW), Clang/LLVM (GNU/Linux, Android, macOS, iOS).
- Build tested with Microsoft Visual Studio 2019, 2022
-1
View File
@@ -344,7 +344,6 @@ int dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_config
config.push_token = conf->push_token ? std::string(conf->push_token) : std::string{};
config.push_topic = conf->push_topic ? std::string(conf->push_topic) : std::string{};
config.push_platform = conf->push_platform ? std::string(conf->push_platform) : std::string{};
config.proxy_user_agent = conf->proxy_user_agent ? std::string(conf->proxy_user_agent) : std::string{};
config.peer_discovery = conf->peer_discovery;
config.peer_publish = conf->peer_publish;
-1
View File
@@ -148,7 +148,6 @@ struct OPENDHT_PUBLIC dht_runner_config {
const char* push_token;
const char* push_topic;
const char* push_platform;
const char* proxy_user_agent;
bool peer_discovery;
bool peer_publish;
dht_certificate* server_ca;
-3
View File
@@ -143,9 +143,6 @@ struct OPENDHT_PUBLIC Config {
* - Larger listen refresh time
*/
bool public_stable {false};
/* Client mode, node will not be used by other nodes to store data. */
bool client_mode {false};
};
/**
+11 -13
View File
@@ -53,21 +53,19 @@ public:
DhtProxyClient();
explicit DhtProxyClient(
std::shared_ptr<crypto::Certificate> serverCA,
crypto::Identity clientIdentity,
std::function<void()> loopSignal,
const std::string& serverHost,
const std::string& userAgent = "",
const std::string& pushClientId = "",
const std::string& pushToken = "",
const std::string& pushTopic = "",
const std::string& pushPlatform = "",
std::shared_ptr<Logger> logger = {}
);
std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
std::function<void()> loopSignal, const std::string& serverHost,
const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
void setHeaderFields(http::Request& request);
virtual void setPushNotificationToken(const std::string& token) override;
virtual void setPushNotificationToken(const std::string& token) override {
#ifdef OPENDHT_PUSH_NOTIFICATIONS
deviceKey_ = token;
#else
(void) token;
#endif
}
virtual void setPushNotificationTopic(const std::string& topic) override {
#ifdef OPENDHT_PUSH_NOTIFICATIONS
@@ -353,7 +351,7 @@ private:
std::string proxyUrl_;
dht::crypto::Identity clientIdentity_;
std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
std::string userAgent_ {"OpenDHT"};
//std::pair<std::string, std::string> serverHostService_;
std::string pushClientId_;
std::string pushSessionId_;
+3 -51
View File
@@ -98,32 +98,6 @@ public:
asio::io_context& io_context() const;
using clock = std::chrono::steady_clock;
using time_point = clock::time_point;
struct PushStats {
uint64_t highPriorityCount {0};
uint64_t normalPriorityCount {0};
void increment(bool highPriority) {
if (highPriority)
highPriorityCount++;
else
normalPriorityCount++;
}
Json::Value toJson() const {
Json::Value val;
val["highPriorityCount"] = static_cast<Json::UInt64>(highPriorityCount);
val["normalPriorityCount"] = static_cast<Json::UInt64>(normalPriorityCount);
return val;
}
std::string toString() const {
return fmt::format("{} high priority, {} normal priority", highPriorityCount, normalPriorityCount);
}
};
struct ServerStats {
/** Current number of listen operations */
size_t listenCount {0};
@@ -133,17 +107,6 @@ public:
size_t totalPermanentPuts {0};
/** Current number of push tokens with at least one listen operation */
size_t pushListenersCount {0};
/** Time at which the server was started */
time_point serverStartTime;
/** Last time at which the stats were updated */
time_point lastUpdated;
/** Total number of push notification requests that the server attempted to
* send since being started, broken down by type and priority level */
PushStats androidPush;
PushStats iosPush;
PushStats unifiedPush;
/** Average requests per second */
double requestRate {0};
/** Node Info **/
@@ -152,10 +115,6 @@ public:
std::string toString() const {
std::ostringstream ss;
ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
ss << "Push requests in the last " << print_duration(lastUpdated - serverStartTime) << ": "
<< "[Android: " << androidPush.toString() << "], "
<< "[iOS: " << iosPush.toString() << "], "
<< "[Unified: " << unifiedPush.toString() << "]" << std::endl;
ss << "Requests: " << requestRate << " per second." << std::endl;
if (nodeInfo) {
auto& ipv4 = nodeInfo->ipv4;
@@ -177,11 +136,6 @@ public:
result["putCount"] = static_cast<Json::UInt64>(putCount);
result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
result["serverStartTime"] = static_cast<Json::LargestInt>(to_time_t(serverStartTime));
result["lastUpdated"] = static_cast<Json::LargestInt>(to_time_t(lastUpdated));
result["androidPush"] = androidPush.toJson();
result["iosPush"] = iosPush.toJson();
result["unifiedPush"] = unifiedPush.toJson();
result["requestRate"] = requestRate;
if (nodeInfo)
result["nodeInfo"] = nodeInfo->toJson();
@@ -412,6 +366,9 @@ private:
template <typename Is>
void loadState(Is& is, size_t size);
using clock = std::chrono::steady_clock;
using time_point = clock::time_point;
std::shared_ptr<asio::io_context> ioContext_;
std::shared_ptr<DhtRunner> dht_;
Json::StreamWriterBuilder jsonBuilder_;
@@ -436,11 +393,6 @@ private:
std::shared_ptr<ServerStats> stats_;
std::shared_ptr<NodeInfo> nodeInfo_ {};
std::unique_ptr<asio::steady_timer> printStatsTimer_;
const time_point serverStartTime_;
mutable std::mutex pushStatsMutex_;
PushStats androidPush_;
PushStats iosPush_;
PushStats unifiedPush_;
// Thread-safe access to listeners map.
std::mutex lockListener_;
+2 -5
View File
@@ -64,7 +64,6 @@ public:
std::string push_token {};
std::string push_topic {};
std::string push_platform {};
std::string proxy_user_agent {};
bool peer_discovery {false};
bool peer_publish {false};
std::shared_ptr<dht::crypto::Certificate> server_ca;
@@ -410,8 +409,7 @@ public:
void run(const Config& config, Context&& context);
void setOnStatusChanged(StatusCallback&& cb) {
if (cb)
statusCbs.emplace_back(std::move(cb));
statusCb = std::move(cb);
}
/**
@@ -521,8 +519,7 @@ private:
NodeStatus status4 {NodeStatus::Disconnected},
status6 {NodeStatus::Disconnected};
std::vector<StatusCallback> statusCbs {};
StatusCallback statusCb {nullptr};
/** PeerDiscovery Parameters */
std::shared_ptr<PeerDiscovery> peerDiscovery_;
-1
View File
@@ -52,7 +52,6 @@ struct NetworkConfig {
NetId network {0};
ssize_t max_req_per_sec {0};
ssize_t max_peer_req_per_sec {0};
bool is_client {false};
};
class DhtProtocolException : public DhtException {
+14 -21
View File
@@ -23,9 +23,7 @@
#include "sockaddr.h"
#include "infohash.h"
#include "logger.h"
#include "utils.h"
#include <asio/steady_timer.hpp>
#include <thread>
namespace asio {
@@ -38,34 +36,31 @@ class OPENDHT_PUBLIC PeerDiscovery
{
public:
static constexpr in_port_t DEFAULT_PORT = 8888;
using ServiceDiscoveredCallback = std::function<void(msgpack::object &&, SockAddr &&)>;
using ServiceDiscoveredCallback = std::function<void(msgpack::object&&, SockAddr&&)>;
PeerDiscovery(in_port_t port = DEFAULT_PORT,
std::shared_ptr<asio::io_context> ioContext = {},
std::shared_ptr<Logger> logger = {});
PeerDiscovery(in_port_t port = DEFAULT_PORT, std::shared_ptr<asio::io_context> ioContext = {}, std::shared_ptr<Logger> logger = {});
~PeerDiscovery();
/**
* startDiscovery - Keep Listening data from the sender until node is joinned
* or stop is called
*/
* startDiscovery - Keep Listening data from the sender until node is joinned or stop is called
*/
void startDiscovery(const std::string &type, ServiceDiscoveredCallback callback);
template <typename T>
void startDiscovery(const std::string &type, std::function<void(T &&, SockAddr &&)> cb) {
startDiscovery(type, [cb](msgpack::object &&ob, SockAddr &&addr) {
template<typename T>
void startDiscovery(const std::string &type, std::function<void(T&&, SockAddr&&)> cb) {
startDiscovery(type, [cb](msgpack::object&& ob, SockAddr&& addr) {
cb(ob.as<T>(), std::move(addr));
});
}
/**
* startPublish - Keeping sending data until node is joinned or stop is called
*/
*/
void startPublish(const std::string &type, const msgpack::sbuffer &pack_buf);
void startPublish(sa_family_t domain, const std::string &type, const msgpack::sbuffer &pack_buf);
template <typename T>
void startPublish(const std::string &type, const T &object) {
template<typename T>
void startPublish(const std::string &type, const T& object) {
msgpack::sbuffer buf;
msgpack::pack(buf, object);
startPublish(type, buf);
@@ -73,24 +68,22 @@ public:
/**
* Thread Stopper
*/
*/
void stop();
/**
* Remove possible callBack to discovery
*/
*/
bool stopDiscovery(const std::string &type);
/**
* Remove different serivce message to send
*/
*/
bool stopPublish(const std::string &type);
bool stopPublish(sa_family_t domain, const std::string &type);
void connectivityChanged();
void stopConnectivityChanged();
private:
class DomainPeerDiscovery;
std::unique_ptr<DomainPeerDiscovery> peerDiscovery4_;
@@ -99,4 +92,4 @@ private:
std::thread ioRunnner_;
};
} // namespace dht
}
+4 -8
View File
@@ -13,16 +13,10 @@ openssl = dependency('openssl', required: get_option('proxy_client'))
jsoncpp = dependency('jsoncpp', required: get_option('proxy_client'))
fmt = dependency('fmt')
llhttp = dependency('llhttp', 'libllhttp', required: get_option('proxy_client'))
io_uring = dependency('liburing', required: false)
deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp, io_uring]
conf_data = configuration_data()
deps = [fmt, gnutls, nettle, msgpack, argon2, openssl, jsoncpp, llhttp]
add_project_arguments('-DMSGPACK_NO_BOOST', '-DASIO_STANDALONE', language : 'cpp')
if io_uring.found()
add_project_arguments('-DASIO_HAS_IO_URING', '-DASIO_DISABLE_EPOLL', language : 'cpp')
conf_data.set('iouring_lib', ', liburing')
endif
add_project_arguments('-DMSGPACK_NO_BOOST', language : 'cpp')
add_project_arguments(['-Wno-return-type','-Wno-deprecated','-Wnon-virtual-dtor','-pedantic-errors','-fvisibility=hidden'], language : 'cpp')
opendht_inc = include_directories('include/opendht')
@@ -46,6 +40,8 @@ opendht_src = [
'src/thread_pool.cpp',
]
conf_data = configuration_data()
if get_option('indexation').enabled()
opendht_src += 'src/indexation/pht.cpp'
add_project_arguments('-DOPENDHT_INDEXATION', language : 'cpp')
+1 -1
View File
@@ -8,5 +8,5 @@ Version: @VERSION@
Libs: -L${libdir} -lopendht
Libs.private: @http_lib@ -pthread
Requires: gnutls >= 3.3@jsoncpp_lib@@openssl_lib@
Requires.private: nettle >= 2.4@argon2_lib@@iouring_lib@
Requires.private: nettle >= 2.4@argon2_lib@
Cflags: -I${includedir}
+1 -2
View File
@@ -500,13 +500,12 @@ cdef class DhtConfig(object):
def setRateLimit(self, ssize_t max_req_per_sec, ssize_t max_peer_req_per_sec):
self._config.dht_config.node_config.max_req_per_sec = max_req_per_sec
self._config.dht_config.node_config.max_peer_req_per_sec = max_peer_req_per_sec
def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", str user_agent="", Certificate server_ca = Certificate()):
def setProxyInfo(self, str proxy_server, str push_node_id="", str push_token="", str push_topic="", str push_platform="", Certificate server_ca = Certificate()):
self._config.proxy_server = proxy_server.encode()
self._config.push_node_id = push_node_id.encode()
self._config.push_token = push_token.encode()
self._config.push_topic = push_topic.encode()
self._config.push_platform = push_platform.encode()
self._config.proxy_user_agent = user_agent.encode()
self._config.server_ca = server_ca._cert
def setPeerDiscovery(self, bool peer_discovery, bool peer_publish):
self._config.peer_discovery = peer_discovery
-1
View File
@@ -246,7 +246,6 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
string push_token
string push_topic
string push_platform
string proxy_user_agent
bool peer_discovery
bool peer_publish
shared_ptr[Certificate] server_ca
+5 -5
View File
@@ -2,10 +2,10 @@
name = "opendht"
version = "0.1.0"
authors = ["Sébastien Blin <sebastien.blin@savoirfairelinux.com>"]
edition = "2021"
[dependencies]
libc="0.2"
os_socketaddr="0.2.3"
edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
libc="0.2.0"
os_socketaddr="0.2.3"
+8 -14
View File
@@ -49,7 +49,6 @@ impl DhtRunnerConfig {
push_token: ptr::null(),
push_topic: ptr::null(),
push_platform: ptr::null(),
proxy_user_agent: ptr::null(),
peer_discovery: false,
peer_publish: false,
server_ca: ptr::null_mut(),
@@ -57,7 +56,6 @@ impl DhtRunnerConfig {
privatekey: ptr::null_mut(),
certificate: ptr::null_mut(),
},
log: true,
});
unsafe {
dht_runner_config_default(&mut *config);
@@ -66,27 +64,23 @@ impl DhtRunnerConfig {
}
pub fn set_proxy_server(&mut self, proxy_server: &str) {
self.proxy_server = CString::new(proxy_server).unwrap().into_raw();
self.proxy_server = CString::new(proxy_server).unwrap().as_ptr();
}
pub fn set_push_node_id(&mut self, push_node_id: &str) {
self.push_node_id = CString::new(push_node_id).unwrap().into_raw();
self.push_node_id = CString::new(push_node_id).unwrap().as_ptr();
}
pub fn set_push_token(&mut self, push_token: &str) {
self.push_token = CString::new(push_token).unwrap().into_raw();
self.push_token = CString::new(push_token).unwrap().as_ptr();
}
pub fn set_push_topic(&mut self, push_topic: &str) {
self.push_topic = CString::new(push_topic).unwrap().into_raw();
self.push_topic = CString::new(push_topic).unwrap().as_ptr();
}
pub fn set_push_platform(&mut self, push_platform: &str) {
self.push_platform = CString::new(push_platform).unwrap().into_raw();
}
pub fn set_proxy_user_agent(&mut self, proxy_user_agent: &str) {
self.proxy_user_agent = CString::new(proxy_user_agent).unwrap().into_raw();
self.push_platform = CString::new(push_platform).unwrap().as_ptr();
}
pub fn set_identity(&mut self, certificate: Box<DhtCertificate>, privatekey: Box<PrivateKey>) {
@@ -99,7 +93,7 @@ impl DhtRunnerConfig {
impl DhtNodeConfig
{
pub fn set_persist_path(&mut self, persist_path: &str) {
self.persist_path = CString::new(persist_path).unwrap().into_raw();
self.persist_path = CString::new(persist_path).unwrap().as_ptr();
}
}
@@ -203,8 +197,8 @@ impl DhtRunner {
pub fn bootstrap(&mut self, host: &str, service: u16) {
unsafe {
dht_runner_bootstrap(&mut *self,
CString::new(host).unwrap().into_raw(),
CString::new(service.to_string()).unwrap().into_raw())
CString::new(host).unwrap().as_ptr(),
CString::new(service.to_string()).unwrap().as_ptr())
}
}
+1 -2
View File
@@ -120,14 +120,13 @@ pub struct DhtRunnerConfig
pub push_token: *const c_char,
pub push_topic: *const c_char,
pub push_platform: *const c_char,
pub proxy_user_agent: *const c_char,
pub peer_discovery: bool,
pub peer_publish: bool,
pub server_ca: *mut DhtCertificate,
pub client_identity: DhtIdentity,
pub log: bool,
}
#[link(name = "opendht-c")]
extern {
// dht::Value
-39
View File
@@ -1,39 +0,0 @@
//Code from getifaddrs.3 man
#include <arpa/inet.h>
#include <sys/socket.h>
#include <ifaddrs.h>
#include <stdlib.h>
#include <unistd.h>
#include <net/if.h>
#include <asio.hpp>
#include <stdexcept>
namespace dht{
namespace workaround{
asio::ip::address_v4
get_interface(){
struct ifaddrs *ifaddr;
if(getifaddrs(&ifaddr) == -1){
throw std::runtime_error("Can't getifaddrs");
}
for (struct ifaddrs *if1 = ifaddr; if1 != NULL; if1 = if1->ifa_next){
if (if1->ifa_addr == NULL)
continue;
if (if1->ifa_addr->sa_family != AF_INET)
continue;
if((if1->ifa_flags & IFF_MULTICAST) == IFF_MULTICAST){
auto sa_data = reinterpret_cast<uint8_t*>(if1->ifa_addr->sa_data);
uint32_t value = sa_data[2] << 24
| sa_data[3] << 16
| sa_data[4] << 8
| sa_data[5];
freeifaddrs(ifaddr);
return asio::ip::address_v4(value);
}
}
freeifaddrs(ifaddr);
throw std::runtime_error("Can't find a interface which supports multicast");
}
} /* namespace workaround */
} /* namespace dht */
+22 -83
View File
@@ -93,27 +93,11 @@ Blob aesEncrypt(const uint8_t* data, size_t data_length, const Blob& key)
std::random_device rdev;
std::generate_n(ret.begin(), GCM_IV_SIZE, std::bind(rand_byte, std::ref(rdev)));
}
if (key.size() == AES_LENGTHS[0]) {
struct gcm_aes128_ctx aes;
gcm_aes128_set_key(&aes, key.data());
gcm_aes128_set_iv(&aes, GCM_IV_SIZE, ret.data());
gcm_aes128_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
gcm_aes128_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
} else if (key.size() == AES_LENGTHS[1]) {
struct gcm_aes192_ctx aes;
gcm_aes192_set_key(&aes, key.data());
gcm_aes192_set_iv(&aes, GCM_IV_SIZE, ret.data());
gcm_aes192_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
gcm_aes192_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
} else if (key.size() == AES_LENGTHS[2]) {
struct gcm_aes256_ctx aes;
gcm_aes256_set_key(&aes, key.data());
gcm_aes256_set_iv(&aes, GCM_IV_SIZE, ret.data());
gcm_aes256_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
gcm_aes256_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
}
struct gcm_aes_ctx aes;
gcm_aes_set_key(&aes, key.size(), key.data());
gcm_aes_set_iv(&aes, GCM_IV_SIZE, ret.data());
gcm_aes_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
gcm_aes_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
return ret;
}
@@ -134,28 +118,14 @@ Blob aesDecrypt(const uint8_t* data, size_t data_length, const Blob& key)
std::array<uint8_t, GCM_DIGEST_SIZE> digest;
struct gcm_aes_ctx aes;
gcm_aes_set_key(&aes, key.size(), key.data());
gcm_aes_set_iv(&aes, GCM_IV_SIZE, data);
size_t data_sz = data_length - GCM_IV_SIZE - GCM_DIGEST_SIZE;
Blob ret(data_sz);
if (key.size() == AES_LENGTHS[0]) {
struct gcm_aes128_ctx aes;
gcm_aes128_set_key(&aes, key.data());
gcm_aes128_set_iv(&aes, GCM_IV_SIZE, data);
gcm_aes128_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
gcm_aes128_digest(&aes, GCM_DIGEST_SIZE, digest.data());
} else if (key.size() == AES_LENGTHS[1]) {
struct gcm_aes192_ctx aes;
gcm_aes192_set_key(&aes, key.data());
gcm_aes192_set_iv(&aes, GCM_IV_SIZE, data);
gcm_aes192_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
gcm_aes192_digest(&aes, GCM_DIGEST_SIZE, digest.data());
} else if (key.size() == AES_LENGTHS[2]) {
struct gcm_aes256_ctx aes;
gcm_aes256_set_key(&aes, key.data());
gcm_aes256_set_iv(&aes, GCM_IV_SIZE, data);
gcm_aes256_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
gcm_aes256_digest(&aes, GCM_DIGEST_SIZE, digest.data());
}
gcm_aes_decrypt(&aes, data_sz, ret.data(), data + GCM_IV_SIZE);
gcm_aes_digest(&aes, GCM_DIGEST_SIZE, digest.data());
if (not std::equal(digest.begin(), digest.end(), data + data_length - GCM_DIGEST_SIZE)) {
throw DecryptError("Can't decrypt data");
@@ -257,7 +227,7 @@ PrivateKey::PrivateKey(const uint8_t* src, size_t src_size, const char* password
throw CryptoException("Can't initialize private key !");
const gnutls_datum_t dt {(uint8_t*)src, static_cast<unsigned>(src_size)};
int flags = (password_ptr == NULL || strlen(password_ptr)==0) ? GNUTLS_PKCS_PLAIN
int flags = password_ptr ? GNUTLS_PKCS_PLAIN
: ( GNUTLS_PKCS_PBES2_AES_128 | GNUTLS_PKCS_PBES2_AES_192 | GNUTLS_PKCS_PBES2_AES_256
| GNUTLS_PKCS_PKCS12_3DES | GNUTLS_PKCS_PKCS12_ARCFOUR | GNUTLS_PKCS_PKCS12_RC2_40);
@@ -1164,17 +1134,11 @@ saveIdentity(const Identity& id, const std::string& path, const std::string& pri
auto ca_key_data = id.first->serialize(privkey_password);
std::ofstream key_file(path + ".pem");
key_file.write((char*)ca_key_data.data(), ca_key_data.size());
// Throw error if the file is not written
if (!key_file)
throw CryptoException("Could not write private key file");
}
{
auto ca_key_data = id.second->getPacked();
std::ofstream crt_file(path + ".crt");
crt_file.write((char*)ca_key_data.data(), ca_key_data.size());
// Throw error if the file is not written
if (!crt_file)
throw CryptoException("Could not write certificate file");
}
}
@@ -1188,19 +1152,12 @@ loadIdentity(const std::string &path,const std::string &privkey_password)
pkStream.close();
// Create a certificate
gnutls_x509_crt_t gnuCert;
if (gnutls_x509_crt_init(&gnuCert))
if (gnutls_x509_crt_init(&gnuCert) != GNUTLS_E_SUCCESS)
throw std::runtime_error("Failed to initialize gnutls certificate struct");
gnutls_datum_t crtContent {nullptr, 0};
// Read the certificate file
int err = gnutls_load_file((path + ".crt").c_str(), &crtContent);
if (err)
throw CryptoException(gnutls_strerror(err));
err = gnutls_x509_crt_import(gnuCert, &crtContent, GNUTLS_X509_FMT_PEM);
if (err)
throw CryptoException(gnutls_strerror(err));
gnutls_load_file((path + ".crt").c_str(), &crtContent);
gnutls_x509_crt_import(gnuCert, &crtContent, GNUTLS_X509_FMT_PEM);
auto cert = std::make_shared<Certificate>(gnuCert);
return {std::move(key), std::move(cert)};
}
@@ -1247,49 +1204,31 @@ Certificate::generate(const PrivateKey& key, const std::string& name, const Iden
auto pk_id = pk.getId();
const std::string uid_str = pk_id.toString();
int err = gnutls_x509_crt_set_subject_key_id(cert, &pk_id, sizeof(pk_id));
if(err) {
throw CryptoException(std::string("Error when setting subject key id ") + gnutls_strerror(err));
}
err = gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_X520_COMMON_NAME, 0, name.data(), name.length());
if(err) {
throw CryptoException(std::string("Error when setting subject key id ") + gnutls_strerror(err));
}
err = gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_LDAP_UID, 0, uid_str.data(), uid_str.length());
if(err) {
throw CryptoException(std::string("Error when setting dn by oid ") + gnutls_strerror(err));
}
gnutls_x509_crt_set_subject_key_id(cert, &pk_id, sizeof(pk_id));
gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_X520_COMMON_NAME, 0, name.data(), name.length());
gnutls_x509_crt_set_dn_by_oid(cert, GNUTLS_OID_LDAP_UID, 0, uid_str.data(), uid_str.length());
setRandomSerial(cert);
unsigned key_usage = 0;
if (is_ca) {
err = gnutls_x509_crt_set_ca_status(cert, 1);
if(err) {
throw CryptoException(std::string("Error when setting ca status ") + gnutls_strerror(err));
}
gnutls_x509_crt_set_ca_status(cert, 1);
key_usage |= GNUTLS_KEY_KEY_CERT_SIGN | GNUTLS_KEY_CRL_SIGN;
} else {
key_usage |= GNUTLS_KEY_DIGITAL_SIGNATURE | GNUTLS_KEY_DATA_ENCIPHERMENT;
}
err = gnutls_x509_crt_set_key_usage(cert, key_usage);
if(err) {
throw CryptoException(std::string("Error when setting ca status ") + gnutls_strerror(err));
}
gnutls_x509_crt_set_key_usage(cert, key_usage);
if (ca.first && ca.second) {
if (not ca.second->isCA()) {
throw CryptoException("Signing certificate must be CA");
}
err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0);
if (err) {
if (int err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, pk.getPreferredDigest(), 0)) {
throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
}
ret.issuer = ca.second;
} else {
err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0);
if (err) {
if (int err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0)) {
throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
}
}
+1 -2
View File
@@ -93,7 +93,7 @@ Dht::shutdown(ShutdownCallback cb, bool stop)
r.second.done_cb(false, {});
sr.second->callbacks.clear();
for (const auto& a : sr.second->announce) {
for (auto& cb : a.callbacks) cb(false, {});
if (a.callback) a.callback(false, {});
}
sr.second->announce.clear();
sr.second->listeners.clear();
@@ -1812,7 +1812,6 @@ fromDhtConfig(const Config& config)
netConf.max_peer_req_per_sec = config.max_peer_req_per_sec
? config.max_peer_req_per_sec
: netConf.max_req_per_sec/8;
netConf.is_client = config.client_mode;
return netConf;
}
+7 -42
View File
@@ -106,26 +106,13 @@ getRandomSessionId(size_t length = 8) {
DhtProxyClient::DhtProxyClient() {}
DhtProxyClient::DhtProxyClient(
std::shared_ptr<dht::crypto::Certificate> serverCA,
dht::crypto::Identity clientIdentity,
std::function<void()> signal,
const std::string& serverHost,
const std::string& userAgent,
const std::string& pushClientId,
const std::string& pushToken,
const std::string& pushTopic,
const std::string& pushPlatform,
std::shared_ptr<dht::Logger> logger
)
std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
std::function<void()> signal, const std::string& serverHost,
const std::string& pushClientId, std::shared_ptr<dht::Logger> logger)
: DhtInterface(logger)
, proxyUrl_(serverHost)
, clientIdentity_(clientIdentity), serverCertificate_(serverCA)
, userAgent_(userAgent)
, pushClientId_(pushClientId)
, pushSessionId_(getRandomSessionId())
, deviceKey_(pushToken)
, notificationTopic_(pushTopic)
, platform_(pushPlatform)
, pushClientId_(pushClientId), pushSessionId_(getRandomSessionId())
, loopSignal_(signal)
, jsonReader_(Json::CharReaderBuilder{}.newCharReader())
{
@@ -497,7 +484,7 @@ DhtProxyClient::buildRequest(const std::string& target)
request->set_certificate_authority(serverCertificate_);
if (clientIdentity_.first and clientIdentity_.second)
request->set_identity(clientIdentity_);
request->set_header_field(restinio::http_field_t::user_agent, userAgent_);
request->set_header_field(restinio::http_field_t::user_agent, "RESTinio client");
return request;
}
@@ -1121,12 +1108,10 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
std::lock_guard<std::mutex> l(requestLock_);
requests_[reqid] = request;
}
request->add_on_status_callback([r=std::weak_ptr(request), seconds = this->listenKeepIdle()] (unsigned status_code) {
request->add_on_status_callback([request, seconds = this->listenKeepIdle()] (unsigned status_code) {
if(status_code == 200) {
// increase TCP_KEEPIDLE to save power
if (auto request = r.lock()) {
request->get_connection()->set_keepalive(seconds);
}
request->get_connection()->set_keepalive(seconds);
}
});
request->send();
@@ -1396,24 +1381,4 @@ DhtProxyClient::fillBody(bool resubscribe)
}
#endif // OPENDHT_PUSH_NOTIFICATIONS
void
DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& token) {
#ifdef OPENDHT_PUSH_NOTIFICATIONS
std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
if (deviceKey_ != token) {
deviceKey_ = token;
if (statusIpv4_ == NodeStatus::Connected || statusIpv6_ == NodeStatus::Connected) {
if (logger_)
logger_->d("[proxy:client] [push] token changed, resubscribing");
for (auto& search : searches_) {
for (auto& listener : search.second.listeners) {
resubscribe(search.first, listener.first, listener.second);
}
}
}
}
#endif
}
} // namespace dht
+4 -32
View File
@@ -223,7 +223,6 @@ DhtProxyServer::DhtProxyServer(const std::shared_ptr<DhtRunner>& dht,
: ioContext_(std::make_shared<asio::io_context>()),
dht_(dht), persistPath_(config.persistStatePath), logger_(logger),
printStatsTimer_(std::make_unique<asio::steady_timer>(*ioContext_, 3s)),
serverStartTime_(clock::now()),
connListener_(std::make_shared<ConnectionListener>(std::bind(&DhtProxyServer::onConnectionClosed, this, std::placeholders::_1))),
pushServer_(config.pushServer),
bundleId_(config.bundleId)
@@ -543,15 +542,7 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
stats.requestRate = count / dt.count();
#ifdef OPENDHT_PUSH_NOTIFICATIONS
stats.pushListenersCount = pushListeners_.size();
{
std::lock_guard lk(pushStatsMutex_);
stats.androidPush = androidPush_;
stats.iosPush = iosPush_;
stats.unifiedPush = unifiedPush_;
}
#endif
stats.serverStartTime = serverStartTime_;
stats.lastUpdated = now;
stats.totalPermanentPuts = 0;
std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
stats.totalPermanentPuts += put.second.puts.size();
@@ -1116,14 +1107,10 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
notification["tokens"] = std::move(tokens);
notification["platform"] = type == PushType::Android ? 2 : 1;
notification["data"] = std::move(json);
auto priority = highPriority ? "high" : "normal";
if (type == PushType::Android) {
Json::Value androidConfig(Json::objectValue);
androidConfig["priority"] = priority;
androidConfig["ttl"] = "86400s"; // time to live = 24 hours
notification["android"] = std::move(androidConfig);
} else {
notification["priority"] = priority;
notification["priority"] = highPriority ? "high" : "normal";
if (type == PushType::Android)
notification["time_to_live"] = 3600 * 24; // 24 hours
else {
const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
notification["expiration"] = exp;
@@ -1164,21 +1151,6 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
requests_[reqid] = request;
}
request->send();
// For monitoring purposes
std::lock_guard lk(pushStatsMutex_);
switch (type) {
case PushType::Android:
androidPush_.increment(highPriority);
break;
case PushType::iOS:
iosPush_.increment(highPriority);
break;
case PushType::UnifiedPush:
unifiedPush_.increment(highPriority);
break;
default:
break;
}
}
catch (const std::exception &e){
if (logger_)
+12 -24
View File
@@ -180,9 +180,9 @@ DhtRunner::run(const Config& config, Context&& context)
throw;
}
statusCbs.clear();
if (context.statusChangedCallback)
statusCbs.emplace_back(std::move(context.statusChangedCallback));
if (context.statusChangedCallback) {
statusCb = std::move(context.statusChangedCallback);
}
if (context.certificateStore) {
dht_->setLocalCertificateStore(std::move(context.certificateStore));
}
@@ -264,17 +264,6 @@ DhtRunner::run(const Config& config, Context&& context)
}
}
}
if (peerDiscovery_) {
statusCbs.emplace_back([this](NodeStatus status4, NodeStatus status6) {
if (status4 == NodeStatus::Disconnected && status6 == NodeStatus::Disconnected) {
peerDiscovery_->connectivityChanged();
}
else if (status4 != NodeStatus::Connected || status6 != NodeStatus::Connected) {
peerDiscovery_->stopConnectivityChanged();
}
});
}
#endif
}
}
@@ -701,9 +690,8 @@ DhtRunner::loop_()
if (nstatus4 != status4 || nstatus6 != status6) {
status4 = nstatus4;
status6 = nstatus6;
for (auto& cb : statusCbs){
cb(status4, status6);
}
if (statusCb)
statusCb(status4, status6);
}
return wakeup;
@@ -1120,13 +1108,13 @@ DhtRunner::enableProxy(bool proxify)
cv.notify_all();
}
},
config_.proxy_server,
config_.proxy_user_agent,
config_.push_node_id,
config_.push_token,
config_.push_topic,
config_.push_platform,
logger_);
config_.proxy_server, config_.push_node_id, logger_);
if (not config_.push_token.empty())
dht_via_proxy->setPushNotificationToken(config_.push_token);
if (not config_.push_topic.empty())
dht_via_proxy->setPushNotificationTopic(config_.push_topic);
if (not config_.push_platform.empty())
dht_via_proxy->setPushNotificationPlatform(config_.push_platform);
dht_ = std::make_unique<SecureDht>(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_);
// and use it
use_proxy = proxify;
+1 -1
View File
@@ -461,7 +461,7 @@ Connection::set_ssl_verification(const std::string& hostname, const asio::ssl::v
bool verified = verifier(preverified, ctx);
auto verify_ec = X509_STORE_CTX_get_error(ctx.native_handle());
if (verify_ec != 0 /*X509_V_OK*/ and logger)
logger->error("[http::connection:{:d}] ssl verification error={:d} {}", id, verify_ec, verified);
logger->error("[http::connection:{:d}] ssl verification error={:d} %d", id, verify_ec, verified);
if (verified and checkOcsp) {
std::unique_ptr<stack_st_X509, void(*)(stack_st_X509*)> chain(
X509_STORE_CTX_get1_chain(ctx.native_handle()),
+16 -61
View File
@@ -143,7 +143,7 @@ NetworkEngine::tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const Inf
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));
pk.pack(version >= 1 ? KEY_A : KEY_U);
pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
@@ -166,9 +166,6 @@ NetworkEngine::tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const Inf
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
if (version >= 1) {
Tid tid (n->getNewTid());
@@ -197,7 +194,7 @@ NetworkEngine::tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoH
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));
pk.pack(version >= 1 ? KEY_A : KEY_U);
pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
@@ -220,9 +217,6 @@ NetworkEngine::tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoH
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
if (version >= 1) {
Tid tid (n->getNewTid());
@@ -710,7 +704,7 @@ NetworkEngine::sendPing(const Sp<Node>& node, RequestCb&& on_done, RequestExpire
Tid tid (node->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
pk.pack(KEY_A); pk.pack_map(1);
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -722,9 +716,6 @@ NetworkEngine::sendPing(const Sp<Node>& node, RequestCb&& on_done, RequestExpire
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::Ping, tid, node,
Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -750,7 +741,7 @@ void
NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4+(config.network?1:0));
pk.pack(KEY_R); pk.pack_map(2);
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -762,9 +753,6 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
send(addr, buffer.data(), buffer.size());
}
@@ -775,7 +763,7 @@ NetworkEngine::sendFindNode(const Sp<Node>& n, const InfoHash& target, want_t wa
Tid tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
pk.pack(KEY_A); pk.pack_map(2 + (want>0?1:0));
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -794,9 +782,6 @@ NetworkEngine::sendFindNode(const Sp<Node>& n, const InfoHash& target, want_t wa
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::FindNode, tid, n,
Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -823,7 +808,7 @@ NetworkEngine::sendGetValues(const Sp<Node>& n, const InfoHash& info_hash, const
Tid tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
unsigned sendQuery = (not query.where.empty() or not query.select.empty()) ? 1 : 0;
unsigned sendWant = (want > 0) ? 1 : 0;
@@ -850,10 +835,7 @@ NetworkEngine::sendGetValues(const Sp<Node>& n, const InfoHash& info_hash, const
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::GetValues, tid, n,
Blob(buffer.data(), buffer.data() + buffer.size()),
[=](const Request& req_status, ParsedMessage&& msg) { /* on done */
@@ -970,13 +952,10 @@ NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const Soc
end = std::min(start + MTU, v.size());
buffer.clear();
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(3+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(3+(config.network?1:0));
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
pk.pack(KEY_Y); pk.pack(KEY_V);
pk.pack(KEY_TID); pk.pack(tid);
pk.pack(KEY_V); pk.pack_map(1);
@@ -997,7 +976,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
{
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4+(config.network?1:0));
pk.pack(KEY_R);
pk.pack_map(2 + (not st.empty()?1:0) + (nodes.size()>0?1:0) + (nodes6.size()>0?1:0) + (not token.empty()?1:0));
@@ -1039,9 +1018,6 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
// send response
send(addr, buffer.data(), buffer.size());
@@ -1113,7 +1089,7 @@ NetworkEngine::sendListen(const Sp<Node>& n,
Tid tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
auto has_query = not query.where.empty() or not query.select.empty();
pk.pack(KEY_A); pk.pack_map(5 + has_query);
@@ -1133,9 +1109,6 @@ NetworkEngine::sendListen(const Sp<Node>& n,
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::Listen, tid, n,
Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -1157,7 +1130,7 @@ void
NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4+(config.network?1:0));
pk.pack(KEY_R); pk.pack_map(2);
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -1169,9 +1142,6 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
send(addr, buffer.data(), buffer.size());
}
@@ -1188,7 +1158,7 @@ NetworkEngine::sendAnnounceValue(const Sp<Node>& n,
Tid tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
bool add_created = created < scheduler.time();
pk.pack(KEY_A); pk.pack_map(add_created ? 5 : 4);
@@ -1208,9 +1178,6 @@ NetworkEngine::sendAnnounceValue(const Sp<Node>& n,
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid, n,
Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -1277,7 +1244,7 @@ NetworkEngine::sendUpdateValues(const Sp<Node>& n,
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 7 : 6));
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -1298,9 +1265,6 @@ NetworkEngine::sendUpdateValues(const Sp<Node>& n,
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -1325,7 +1289,7 @@ NetworkEngine::sendRefreshValue(const Sp<Node>& n,
Tid tid (n->getNewTid());
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(5+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(5+(config.network?1:0));
pk.pack(KEY_A); pk.pack_map(4);
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -1340,9 +1304,6 @@ NetworkEngine::sendRefreshValue(const Sp<Node>& n,
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
auto req = std::make_shared<Request>(MessageType::Refresh, tid, n,
Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -1374,7 +1335,7 @@ void
NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid) {
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4+(config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4+(config.network?1:0));
pk.pack(KEY_R); pk.pack_map(3);
pk.pack(KEY_REQ_ID); pk.pack(myid);
@@ -1387,9 +1348,6 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
send(addr, buffer.data(), buffer.size());
}
@@ -1403,7 +1361,7 @@ NetworkEngine::sendError(const SockAddr& addr,
{
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
pk.pack_map(4 + (include_id?1:0) + (config.network?1:0)+(config.is_client?1:0));
pk.pack_map(4 + (include_id?1:0) + (config.network?1:0));
pk.pack(KEY_E); pk.pack_array(2);
pk.pack(code);
@@ -1420,9 +1378,6 @@ NetworkEngine::sendError(const SockAddr& addr,
if (config.network) {
pk.pack(KEY_NETID); pk.pack(config.network);
}
if (config.is_client) {
pk.pack(KEY_ISCLIENT); pk.pack(config.is_client);
}
send(addr, buffer.data(), buffer.size());
}
+3 -61
View File
@@ -22,10 +22,6 @@
#include "network_utils.h"
#include "utils.h"
#ifdef __ANDROID__
#include "compat/getif_workaround_android.h"
#endif
#include <asio.hpp>
using namespace std::literals;
@@ -52,7 +48,6 @@ public:
void connectivityChanged();
void stopConnectivityChanged();
private:
Sp<Logger> logger_;
//dmtx_ for callbackmap_ and drunning_ (write)
@@ -60,14 +55,6 @@ private:
//mtx_ for messages_ and lrunning (listen)
std::mutex mtx_;
std::shared_ptr<asio::io_context> ioContext_;
static constexpr dht::duration PeerDiscovery_PERIOD_MAX{
std::chrono::minutes(1)};
static constexpr std::chrono::seconds PeerDiscovery_PERIOD{10};
asio::steady_timer peerDiscoveryTimer;
std::chrono::steady_clock::duration peerDiscovery_period{
PeerDiscovery_PERIOD};
asio::ip::udp::socket sockFd_;
asio::ip::udp::endpoint sockAddrSend_;
@@ -95,7 +82,6 @@ private:
PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
: logger_(logger)
, ioContext_(ioContext)
, peerDiscoveryTimer(*ioContext_)
, sockFd_(*ioContext_, domain)
, sockAddrSend_(asio::ip::address::from_string(domain.family() == AF_INET ? MULTICAST_ADDRESS_IPV4
: MULTICAST_ADDRESS_IPV6), port)
@@ -105,23 +91,8 @@ PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(asio::ip::udp domain, in
sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
sockFd_.bind({domain, port});
} catch (const std::exception& e) {
#ifdef __ANDROID__
if(domain.family() == AF_INET && strcmp(e.what(), "set_option: No such device") == 0) {
try{
sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
auto mc_interface = workaround::get_interface();
sockFd_.set_option(asio::ip::multicast::outbound_interface(mc_interface));
sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address().to_v4(), mc_interface));
sockFd_.bind({domain, port});
} catch(const std::exception& e){
if (logger_)
logger_->error("Can't start peer discovery using android workaround: {}", e.what());
}
}
else
#endif
if (logger_)
logger_->error("Can't start peer discovery for {}: {}",
logger_->e("Can't start peer discovery for %s: %s",
domain.family() == AF_INET ? "IPv4" : "IPv6", e.what());
}
}
@@ -327,30 +298,8 @@ PeerDiscovery::DomainPeerDiscovery::reDiscover()
void
PeerDiscovery::DomainPeerDiscovery::connectivityChanged()
{
ioContext_->post([this] () {
reDiscover();
publish(sockAddrSend_);
});
if (logger_)
logger_->d("PeerDiscovery: connectivity changed");
if (peerDiscovery_period == PeerDiscovery_PERIOD_MAX ){
peerDiscovery_period = PeerDiscovery_PERIOD;
}
else{
peerDiscoveryTimer.expires_after(peerDiscovery_period);
peerDiscoveryTimer.async_wait([this](const asio::error_code& ec) {
if (ec == asio::error::operation_aborted)
return;
connectivityChanged();
});
peerDiscovery_period = std::min(peerDiscovery_period * 2, PeerDiscovery_PERIOD_MAX);
}
}
void PeerDiscovery::DomainPeerDiscovery::stopConnectivityChanged() {
peerDiscoveryTimer.cancel();
peerDiscovery_period = PeerDiscovery_PERIOD;
reDiscover();
publish(sockAddrSend_);
}
PeerDiscovery::PeerDiscovery(in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
@@ -463,11 +412,4 @@ PeerDiscovery::connectivityChanged()
peerDiscovery6_->connectivityChanged();
}
void PeerDiscovery::stopConnectivityChanged() {
if (peerDiscovery4_)
peerDiscovery4_->stopConnectivityChanged();
if (peerDiscovery6_)
peerDiscovery6_->stopConnectivityChanged();
}
} /* namespace dht */
+19 -26
View File
@@ -45,7 +45,7 @@ struct Dht::Announce {
bool permanent;
Sp<Value> value;
time_point created;
std::vector<DoneCallback> callbacks;
DoneCallback callback;
};
struct Dht::SearchNode {
@@ -442,9 +442,8 @@ struct Dht::Search {
g.second.done_cb = {};
}
for (auto& a : announce) {
for (auto& cb : a.callbacks)
cb(false, {});
a.callbacks.clear();
a.callback(false, {});
a.callback = {};
}
}
@@ -628,9 +627,7 @@ struct Dht::Search {
return a.value->id == value->id;
});
if (a_sr == announce.end()) {
auto& a = announce.emplace_back(Announce {permanent, value, created, {}} );
if (callback)
a.callbacks.emplace_back(std::move(callback));
announce.emplace_back(Announce {permanent, value, created, callback});
for (auto& n : nodes) {
n->probe_query.reset();
n->acked[value->id].req.reset();
@@ -638,26 +635,24 @@ struct Dht::Search {
} else {
a_sr->permanent = permanent;
a_sr->created = created;
if (a_sr->value != value && *a_sr->value != *value) {
// Value is updated, previous ops are failed
if (a_sr->value != value) {
a_sr->value = value;
for (auto& n : nodes) {
n->acked[value->id].req.reset();
n->probe_query.reset();
}
for (auto& cb: a_sr->callbacks)
cb(false, {});
a_sr->callbacks.clear();
if (callback)
a_sr->callbacks.emplace_back(std::move(callback));
} else if (isAnnounced(value->id)) {
// Same value, already announced
}
if (isAnnounced(value->id)) {
if (a_sr->callback)
a_sr->callback(true, {});
a_sr->callback = {};
if (callback)
callback(true, {});
return;
} else {
// Same value, not announced yet
if (callback)
a_sr->callbacks.emplace_back(std::move(callback));
if (a_sr->callback)
a_sr->callback(false, {});
a_sr->callback = callback;
}
}
}
@@ -727,8 +722,8 @@ struct Dht::Search {
std::vector<DoneCallback> a_cbs;
a_cbs.reserve(announce.size());
for (auto ait = announce.begin() ; ait != announce.end(); ) {
a_cbs.insert(a_cbs.end(), std::make_move_iterator(ait->callbacks.begin()), std::make_move_iterator(ait->callbacks.end()));
ait->callbacks.clear();
if (ait->callback)
a_cbs.emplace_back(std::move(ait->callback));
if (not ait->permanent)
ait = announce.erase(ait);
else
@@ -752,11 +747,9 @@ struct Dht::Search {
if (vid != Value::INVALID_ID and (!a.value || a.value->id != vid))
return true;
if (isAnnounced(a.value->id)) {
if (!a.callbacks.empty()) {
const auto& nodes = getNodes();
for (auto& cb : a.callbacks)
cb(true, nodes);
a.callbacks.clear();
if (a.callback) {
a.callback(true, getNodes());
a.callback = nullptr;
}
if (not a.permanent)
return false;
-1
View File
@@ -271,7 +271,6 @@ Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to)
setRecipient(to.getId());
sign(from);
Value nv {id};
nv.priority = priority;
nv.setCypher(to.encrypt(getToEncrypt()));
return nv;
}
-21
View File
@@ -239,27 +239,6 @@ void CryptoTester::testAesEncryption() {
CPPUNIT_ASSERT(data2 == decrypted2);
}
void CryptoTester::testAesEncryptionWithMultipleKeySizes() {
auto data = std::vector<uint8_t>(rand(), rand());
// Valid key sizes
for (auto key_length : {16, 24, 32}) {
auto key = std::vector<uint8_t>(key_length, rand());
auto encrypted_data = dht::crypto::aesEncrypt(data, key);
auto decrypted_data = dht::crypto::aesDecrypt(encrypted_data, key);
CPPUNIT_ASSERT(data == decrypted_data);
}
// Invalid key sizes
for (auto key_length : {12, 28, 36}) {
auto key = std::vector<uint8_t>(key_length, rand());
CPPUNIT_ASSERT_THROW(dht::crypto::aesEncrypt(data, key), dht::crypto::DecryptError);
}
}
void
CryptoTester::tearDown() {
-2
View File
@@ -34,7 +34,6 @@ class CryptoTester : public CppUnit::TestFixture {
CPPUNIT_TEST(testCertificateSerialNumber);
CPPUNIT_TEST(testOcsp);
CPPUNIT_TEST(testAesEncryption);
CPPUNIT_TEST(testAesEncryptionWithMultipleKeySizes);
CPPUNIT_TEST_SUITE_END();
public:
@@ -70,7 +69,6 @@ class CryptoTester : public CppUnit::TestFixture {
* Test key streching and aes encryption/decryption
*/
void testAesEncryption();
void testAesEncryptionWithMultipleKeySizes();
};
} // namespace test
+1 -54
View File
@@ -90,59 +90,6 @@ DhtRunnerTester::testGetPut() {
CPPUNIT_ASSERT(vals.front()->data == val_data);
}
void
DhtRunnerTester::testPutDuplicate() {
auto key = dht::InfoHash::get("123");
auto val = std::make_shared<dht::Value>("hey");
val->id = 42;
auto val_data = val->data;
std::promise<bool> p1;
std::promise<bool> p2;
node2.put(key, val, [&](bool ok){
p1.set_value(ok);
});
node2.put(key, val, [&](bool ok){
p2.set_value(ok);
});
auto p1ret = p1.get_future().get();
auto p2ret = p2.get_future().get();
CPPUNIT_ASSERT(p1ret);
CPPUNIT_ASSERT(p2ret);
auto vals = node1.get(key).get();
CPPUNIT_ASSERT(not vals.empty());
CPPUNIT_ASSERT(vals.size() == 1);
CPPUNIT_ASSERT(vals.front()->data == val_data);
}
void
DhtRunnerTester::testPutOverride() {
auto key = dht::InfoHash::get("123");
auto val = std::make_shared<dht::Value>("meh");
val->id = 42;
auto val2 = std::make_shared<dht::Value>("hey");
val2->id = 42;
CPPUNIT_ASSERT_EQUAL(val->id, val2->id);
auto val_data = val->data;
std::promise<bool> p1;
std::promise<bool> p2;
node2.put(key, val, [&](bool ok){
p1.set_value(ok);
});
node2.put(key, val2, [&](bool ok){
p2.set_value(ok);
});
auto p1ret = p1.get_future().get();
auto p2ret = p2.get_future().get();
CPPUNIT_ASSERT(!p1ret);
CPPUNIT_ASSERT(p2ret);
auto vals = node1.get(key).get();
CPPUNIT_ASSERT(not vals.empty());
CPPUNIT_ASSERT(vals.size() == 1);
CPPUNIT_ASSERT(vals.front()->data == val_data);
}
void
DhtRunnerTester::testListen() {
std::mutex mutex;
@@ -342,7 +289,7 @@ DhtRunnerTester::testIdOps() {
valueCountEdit++;
cv.notify_all();
});
node2.listen(key2, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool /*expired*/){
node2.listen(key2, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired){
for (const auto& v : values) {
if (v->seq == 0)
CPPUNIT_ASSERT_EQUAL("v1"s, dht::unpackMsg<std::string>(v->data));
-9
View File
@@ -31,7 +31,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
CPPUNIT_TEST_SUITE(DhtRunnerTester);
CPPUNIT_TEST(testConstructors);
CPPUNIT_TEST(testGetPut);
CPPUNIT_TEST(testPutDuplicate);
CPPUNIT_TEST(testListen);
CPPUNIT_TEST(testListenLotOfBytes);
CPPUNIT_TEST(testIdOps);
@@ -56,14 +55,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
* Test get and put methods
*/
void testGetPut();
/**
* Test get and multiple put
*/
void testPutDuplicate();
/**
* Test get and multiple put with changing value
*/
void testPutOverride();
/**
* Test listen method
*/