Compare commits

...
51 Commits
Author SHA1 Message Date
Adrien Béraud 2dbf57d36b Update README.md 2019-12-16 23:42:31 -05:00
Adrien Béraud 5438d50578 proxy: add session token 2019-12-16 16:37:07 -05:00
Adrien Béraud 548255073f docker: build c bindings, use asio 1.12 2019-12-16 12:22:05 -05:00
Adrien Béraud abaf091772 c binding: fix build 2019-12-16 11:41:49 -05:00
Adrien Béraud 7094b79714 crypto: add PublicKey::pack(uint8_t* out, size_t* out_len) 2019-12-14 15:29:42 -05:00
Adrien Béraud 1f672773ba proxy server: respect value priority for push notifications 2019-12-14 13:02:07 -05:00
Adrien Béraud 374fdd6cc9 proxy: send normal priority push message for value expiration 2019-12-14 13:02:07 -05:00
Adrien Béraud 54b1314ab8 value: optimize json parsing 2019-12-14 13:02:07 -05:00
Adrien Béraud 51c35d8b54 value: add priority 2019-12-14 13:02:07 -05:00
Adrien Béraud fea736428e msgpack: take const reference in unpack 2019-12-14 13:01:24 -05:00
Adrien Béraud 2d358adf0e utils: add findMapValue(string) 2019-12-14 13:00:13 -05:00
Adrien Béraud 26c2e16a42 value: optimize json parsing 2019-12-14 00:41:41 -05:00
Adrien Béraud e8b385c220 utils: optimize findMapValue 2019-12-13 23:57:22 -05:00
Adrien Béraud c1e2a368a8 docker: add ssl, install http-parser to /usr 2019-12-13 16:24:46 -05:00
Adrien Béraud c62f27c929 tools: add missing include 2019-12-12 17:00:35 -05:00
Adrien Béraud ec0c3d4d25 docker: build asio without tests and examples 2019-12-12 15:04:45 -05:00
Adrien Béraud 17a1178054 tools: add missing includes for indexation support 2019-12-12 15:00:17 -05:00
Adrien Béraud 6256c576c9 configure: cleanup 2019-12-12 14:51:59 -05:00
Adrien Béraud 9565adf693 proxy server: always depend on jsoncpp 2019-12-12 14:51:59 -05:00
Adrien Béraud c10aa32596 cpp: include config.h when necessary 2019-12-12 14:51:59 -05:00
Adrien Béraud 21c294f352 docker: update asio 2019-12-12 14:51:59 -05:00
Adrien Béraud 4654f78e08 autotools: allow to build without peer discovery 2019-12-12 14:51:59 -05:00
Adrien Béraud 38644fc934 includes: don't include proxy server, pht by default 2019-12-12 14:44:53 -05:00
Adrien Béraud fa8dc4c835 callbacks: add missing header 2019-12-12 14:33:00 -05:00
Adrien Béraud 6dab10cac2 github: disable tests and examples when building asio 2019-12-12 12:24:56 -05:00
Adrien Béraud 77a1c27f2b github: build asio 2019-12-12 11:52:43 -05:00
Adrien Béraud e8577bfd05 dhtrunner: fix build without peer discovery 2019-12-12 11:42:06 -05:00
Adrien Béraud c0327a9881 dhtrunner: fix build without peer discovery 2019-12-11 11:43:20 -05:00
Adrien Béraud 2a1eacc949 tests: only link with ssl if required 2019-12-09 16:23:15 -05:00
Adrien Béraud e705bebc25 github: don't build proxy 2019-12-09 16:06:39 -05:00
Adrien Béraud a330762f40 build: always use standalone asio 2019-12-09 15:40:48 -05:00
Adrien Béraud e7972ae897 dockerfiles: update 2019-12-09 15:34:26 -05:00
Adrien Béraud 8d2d40cb7f peer discovery: simplify includes 2019-12-09 14:53:27 -05:00
Adrien Béraud aceb548c26 cmake: make peer discovery build optional 2019-12-09 14:43:04 -05:00
Adrien Béraud afce4639e1 tests/peerdiscover: add missing includes 2019-12-09 14:07:42 -05:00
Adrien Béraud e5d248bcd0 peer discovery: use asio 2019-12-09 03:28:45 -05:00
Adrien Béraud 0fbebf8e03 http: cleanup, fix handling error when no logger is set 2019-12-08 10:48:34 -05:00
Adrien Béraud 24efa37846 http: cleanup 2019-12-08 00:25:21 -05:00
Linas Vepstas c0f14ca59d Fix duplicated, non-functional log message.
Inside the if-statement, the log message will always print zero.
Confusingly, its the same message as the one in the loop above it.
Make the message distinct.
2019-12-07 22:28:35 -05:00
Adrien Béraud 05a66cc1f1 proxy client: use common jsonReader 2019-12-07 22:22:37 -05:00
Adrien Béraud ed6df8869c proxy: don't use keep-alive by default 2019-12-07 22:22:37 -05:00
Adrien Béraud b7d410e1aa tests: add http tester 2019-12-07 16:23:08 -05:00
Adrien Béraud 77a377a406 http: add OnDone callback 2019-12-07 16:23:08 -05:00
Adrien Béraud e0b6f152f1 proxy client: add LineSplitter 2019-12-07 16:23:08 -05:00
Adrien Béraud dd3c789916 proxy server: 0-initialize fields 2019-12-07 16:23:08 -05:00
Adrien Béraud 333087bcba dhtrunner: fix getNodeInfo using proxy client 2019-12-07 16:23:08 -05:00
Adrien Béraud 5b87090cab http: use http_parser parsing logic 2019-12-07 16:23:08 -05:00
Adrien Béraud fffee634f1 proxy client: avoid destroying twice 2019-12-05 18:39:39 -05:00
Adrien Béraud c4aa272dbb http/request: add constructor 2019-12-05 18:39:39 -05:00
Adrien Béraud 4ce78bdb4d c wrapper: link statically when required 2019-12-05 18:35:43 -05:00
Adrien Béraud f331f39c77 c wrapper: add dht_privatekey_export 2019-12-05 18:35:43 -05:00
45 changed files with 946 additions and 1109 deletions

No files matched your search

+8 -1
View File
@@ -13,12 +13,19 @@ jobs:
run: |
sudo apt install libncurses5-dev libreadline-dev nettle-dev \
libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libasio-dev libfmt-dev libhttp-parser-dev libmsgpack-dev
autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev
- name: autogen
run: ./autogen.sh
- name: argon2
run: |
cd argon2 && make && sudo make install && cd ..
- name: asio
run: |
wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \
&& tar -xvf b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz && cd asio-b2b7a1c166390459e1c169c8ae9ef3234b361e3f/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests \
&& sudo make install \
&& cd ../../ && rm -rf asio*
- name: configure
run: ./configure
- name: make
+28 -7
View File
@@ -22,6 +22,7 @@ option (OPENDHT_PROXY_SERVER_IDENTITY "Allow clients to use the node identity" O
option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restinio and jsoncpp" OFF)
option (OPENDHT_PROXY_OPENSSL "Build DHT proxy with OpenSSL" ON)
option (OPENDHT_PROXY_HTTP_PARSER_FORK "Build DHT proxy with custom http_parser to support old API" OFF)
option (OPENDHT_PEER_DISCOVERY "Enable multicast peer discovery" ON)
option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
option (OPENDHT_TESTS "Add unit tests executable" OFF)
option (OPENDHT_C "Build C bindings" OFF)
@@ -63,7 +64,6 @@ endif()
if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
find_package(Restinio REQUIRED)
if (Restinio_FOUND)
add_definitions(-DASIO_STANDALONE)
find_library(FMT_LIBRARY fmt)
add_library(fmt SHARED IMPORTED)
find_library(HTTP_PARSER_LIBRARY http_parser)
@@ -85,6 +85,11 @@ if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
if (OPENDHT_PROXY_HTTP_PARSER_FORK)
add_definitions(-DOPENDHT_PROXY_HTTP_PARSER_FORK)
endif()
else()
set(OPENDHT_PROXY_OPENSSL OFF)
endif()
if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT OR OPENDHT_PEER_DISCOVERY)
add_definitions(-DASIO_STANDALONE)
endif()
# Build flags
@@ -177,7 +182,6 @@ list (APPEND opendht_SOURCES
src/securedht.cpp
src/dhtrunner.cpp
src/log.cpp
src/peer_discovery.cpp
src/network_utils.cpp
src/thread_pool.cpp
)
@@ -203,12 +207,17 @@ list (APPEND opendht_HEADERS
include/opendht/securedht.h
include/opendht/log.h
include/opendht/log_enable.h
include/opendht/peer_discovery.h
include/opendht/thread_pool.h
include/opendht/network_utils.h
include/opendht.h
)
if (OPENDHT_PEER_DISCOVERY)
list (APPEND opendht_SOURCES src/peer_discovery.cpp)
list (APPEND opendht_HEADERS include/opendht/peer_discovery.h)
add_definitions(-DOPENDHT_PEER_DISCOVERY)
endif()
if (OPENDHT_PYTHON)
message("Indexation enabled since it is required for Python support")
set(OPENDHT_INDEX ON)
@@ -322,7 +331,12 @@ if (OPENDHT_C)
c/opendht_c.h
)
target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
target_link_libraries(opendht-c opendht)
if (OPENDHT_SHARED)
target_link_libraries(opendht-c opendht)
else ()
target_link_libraries(opendht-c opendht-static)
endif ()
# PkgConfig module
configure_file (
opendht-c.pc.in
@@ -374,8 +388,6 @@ if (OPENDHT_TESTS)
tests/cryptotester.cpp
tests/dhtrunnertester.h
tests/dhtrunnertester.cpp
tests/peerdiscoverytester.h
tests/peerdiscoverytester.cpp
tests/threadpooltester.h
tests/threadpooltester.cpp
)
@@ -387,6 +399,12 @@ if (OPENDHT_TESTS)
tests/dhtproxytester.cpp
)
endif()
if (OPENDHT_PEER_DISCOVERY)
list (APPEND test_FILES
tests/peerdiscoverytester.h
tests/peerdiscoverytester.cpp
)
endif()
add_executable(opendht_unit_tests
tests/tests_runner.cpp
${test_FILES}
@@ -401,8 +419,11 @@ if (OPENDHT_TESTS)
${CMAKE_THREAD_LIBS_INIT}
${CPPUNIT_LIBRARIES}
${GNUTLS_LIBRARIES}
ssl crypto jsoncpp
${Jsoncpp_LIBRARIES}
)
if (OPENDHT_PROXY_OPENSSL)
target_link_libraries(opendht_unit_tests ${OPENSSL_LIBRARIES})
endif()
enable_testing()
add_test(TEST opendht_unit_tests)
endif()
+1 -1
View File
@@ -16,7 +16,7 @@ Values are distributed over the network, with redundancy.
* IPv4 and IPv6 support
* Clean and powerful C++14 map API
* Python 3 bindings
* REST API
* REST API with optional HTTP client+server with push notification support
## Documentation
See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>
+9 -2
View File
@@ -122,9 +122,9 @@ void dht_publickey_delete(dht_publickey* pk) {
delete reinterpret_cast<PubkeySp*>(pk);
}
int dht_publickey_pack(dht_publickey* pk, char* out, size_t* outlen) {
int dht_publickey_export(const dht_publickey* pk, char* out, size_t* outlen) {
const auto& pkey = *reinterpret_cast<const PubkeySp*>(pk);
return gnutls_pubkey_export(pkey->pk, GNUTLS_X509_FMT_DER, out, outlen);
return pkey->pack((uint8_t*)out, outlen);
}
dht_infohash dht_publickey_get_id(const dht_publickey* pk) {
@@ -164,6 +164,13 @@ dht_privatekey* dht_privatekey_import(const uint8_t* dat, size_t dat_size, const
}
}
int dht_privatekey_export(const dht_privatekey* k, char* out, size_t* out_size, const char* password) {
if (!out or !out_size or !*out_size)
return -1;
const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
return key->serialize((uint8_t*)out, out_size, password);
}
dht_publickey* dht_privatekey_get_publickey(const dht_privatekey* k) {
const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<dht::crypto::PublicKey>(key->getPublicKey())));
+2 -1
View File
@@ -45,7 +45,7 @@ struct OPENDHT_C_PUBLIC dht_publickey;
typedef struct dht_publickey dht_publickey;
OPENDHT_C_PUBLIC dht_publickey* dht_publickey_import(const uint8_t* dat, size_t dat_size);
OPENDHT_C_PUBLIC void dht_publickey_delete(dht_publickey* pk);
OPENDHT_C_PUBLIC int dht_publickey_pack(dht_publickey* pk, char* out, size_t* out_size);
OPENDHT_C_PUBLIC int dht_publickey_export(const dht_publickey* pk, char* out, size_t* out_size);
OPENDHT_C_PUBLIC dht_infohash dht_publickey_get_id(const dht_publickey* pk);
OPENDHT_C_PUBLIC dht_pkid dht_publickey_get_long_id(const dht_publickey* pk);
OPENDHT_C_PUBLIC bool dht_publickey_check_signature(const dht_publickey* pk, const char* data, size_t data_size, const char* signature, size_t signature_size);
@@ -56,6 +56,7 @@ struct OPENDHT_C_PUBLIC dht_privatekey;
typedef struct dht_privatekey dht_privatekey;
OPENDHT_C_PUBLIC dht_privatekey* dht_privatekey_generate(unsigned key_length_bits);
OPENDHT_C_PUBLIC dht_privatekey* dht_privatekey_import(const uint8_t* dat, size_t dat_size, const char* password);
OPENDHT_C_PUBLIC int dht_privatekey_export(const dht_privatekey*, char* out, size_t* out_size, const char* password);
OPENDHT_C_PUBLIC dht_publickey* dht_privatekey_get_publickey(const dht_privatekey*);
OPENDHT_C_PUBLIC void dht_privatekey_delete(dht_privatekey*);
+10 -6
View File
@@ -35,10 +35,14 @@ AS_IF([test "x$enable_logs" != "xno"], [
dnl Check for indexation
AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--disable-indexation], [Disable DHT indexation])])
AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation != "xno")
AS_IF([test "x$enable_indexation" != "xno"], [
AC_DEFINE([OPENDHT_INDEXATION], [1], [Define if DHT indexation is enabled])
AM_COND_IF(ENABLE_INDEXATION, [
AC_DEFINE([OPENDHT_INDEXATION], [], [Define if DHT indexation is enabled])
])
AC_ARG_ENABLE([peer-discovery], [AS_HELP_STRING([--disable-peer-discovery], [Disable peer-discovery])])
AM_CONDITIONAL(ENABLE_PEER_DISCOVERY, test x$enable_peer_discovery != "xno")
AM_COND_IF(ENABLE_PEER_DISCOVERY, [AC_DEFINE([OPENDHT_PEER_DISCOVERY], [], [Define if peer discovery is enabled])])
dnl Check for Doxygen
AC_ARG_ENABLE([doc], AS_HELP_STRING([--enable-doc], [Enable documentation generation (doxygen)]))
AS_IF([test "x$enable_doc" = "xyes"], [
@@ -135,7 +139,7 @@ AS_IF([test "x$with_jsoncpp" != "xno"],
[have_jsoncpp=no])
AS_IF([test "x$have_jsoncpp" = "xyes"], [
AC_MSG_NOTICE([Using JsonCpp])
CPPFLAGS+=" -DOPENDHT_JSONCPP=1"
CPPFLAGS+=" -DOPENDHT_JSONCPP"
], [
AC_MSG_NOTICE([Not using JsonCpp])
AM_COND_IF(PROXY_CLIENT_OR_SERVER, AC_MSG_ERROR(["JsonCpp is required for proxy/push notification support"]))
@@ -154,7 +158,7 @@ AS_IF([test "x$have_openssl" = "xyes"], [
AC_ARG_WITH([http_parser_fork], AS_HELP_STRING([--with-http-parser-fork], [Build with http_parser fork to support old API]))
AS_IF([test "x$with_http_parser_fork" = "xyes"],[
AC_MSG_NOTICE([Using http_parser fork])
CPPFLAGS+=" -DOPENDHT_PROXY_HTTP_PARSER_FORK"
AC_DEFINE([OPENDHT_PROXY_HTTP_PARSER_FORK], [], [Define if using http parser fork])
], [
AC_MSG_NOTICE([Not using http_parser fork])
])
@@ -205,8 +209,8 @@ AM_COND_IF([ENABLE_TOOLS], [
])
])
AM_COND_IF(ENABLE_PROXY_SERVER, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER"], [])
AM_COND_IF(ENABLE_PROXY_CLIENT, [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT"], [])
AM_COND_IF(ENABLE_PROXY_SERVER, AC_DEFINE([OPENDHT_PROXY_SERVER], [], [Building with proxy server]))
AM_COND_IF(ENABLE_PROXY_CLIENT, AC_DEFINE([OPENDHT_PROXY_CLIENT], [], [Building with proxy client]))
AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS, [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS"], [])
AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY"], [])
+9 -9
View File
@@ -3,13 +3,13 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
RUN apt-get update && apt-get install -y \
build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libasio-dev \
autotools-dev autoconf libssl-dev \
&& apt-get clean
RUN echo "*** Installing libfmt ***" \
&& wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
&& tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On . \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
&& make -j8 && make install \
&& cd ../ && rm -rf fmt*
@@ -17,22 +17,22 @@ RUN echo "*** Installing libfmt ***" \
RUN echo "*** Building nodejs/http_parser dependency ***" \
&& wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
&& tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
&& PREFIX=/usr make -j8 && make install \
&& PREFIX=/usr make -j8 install \
&& cd ../ && rm -rf http-parser*
# libasio-dev (1.10.6-3) is too old
RUN echo "** Building a recent version of asio ***" \
&& wget https://github.com/chriskohlhoff/asio/archive/asio-1-12-2.tar.gz \
&& tar -xvf asio-1-12-2.tar.gz && cd asio-asio-1-12-2/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost \
&& wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
&& tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests \
&& make install \
&& cd ../../ && rm -rf asio*
RUN echo "*** Downloading RESTinio ***" \
&& mkdir restinio && cd restinio \
&& wget https://github.com/aberaud/restinio/archive/8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
&& ls -l && tar -xzf 8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59.tar.gz \
&& cd restinio-8d5d3e8237e0947adb9ba1ffc8281f4ad7cb2a59/dev \
&& wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
&& ls -l && tar -xzf a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
&& cd restinio-a7a10e419d9089c5b8ee63f5e3098c892f22fae4/dev \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
-DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
-DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
+4 -4
View File
@@ -5,14 +5,14 @@ RUN echo "APT::Acquire::Retries \"3\";" > /etc/apt/apt.conf.d/80-retries
RUN apt-get update && apt-get install -y \
apt-transport-https build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
autotools-dev autoconf libasio-dev libfmt-dev libhttp-parser-dev libmsgpack-dev \
autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libssl-dev \
&& apt-get clean
# libasio-dev (1.10) is too old
RUN echo "** Building a recent version of asio ***" \
&& wget https://github.com/chriskohlhoff/asio/archive/asio-1-12-2.tar.gz \
&& tar -xvf asio-1-12-2.tar.gz && cd asio-asio-1-12-2/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost \
&& wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
&& tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests \
&& make install \
&& cd ../../ && rm -rf asio*
+10 -10
View File
@@ -2,9 +2,9 @@ FROM ubuntu:16.04
MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
RUN apt-get update \
&& apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev \
nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev libasio-dev cython3 python3-dev \
nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
python3-setuptools libcppunit-dev python3-pip \
autotools-dev autoconf libasio-dev \
autotools-dev autoconf libssl-dev \
&& apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean
ENV CC cc
@@ -13,7 +13,7 @@ ENV CXX c++
RUN echo "*** Installing libfmt ***" \
&& wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
&& tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On . \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
&& make -j8 && make install \
&& cd ../ && rm -rf fmt*
@@ -21,14 +21,14 @@ RUN echo "*** Installing libfmt ***" \
RUN echo "*** Building nodejs/http_parser dependency ***" \
&& wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
&& tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
&& PREFIX=/usr make -j8 && make install \
&& PREFIX=/usr make -j8 install \
&& cd ../ && rm -rf http-parser*
# libasio-dev (1.10.6-3) is too old
RUN echo "** Building a recent version of asio ***" \
&& wget https://github.com/chriskohlhoff/asio/archive/asio-1-12-2.tar.gz \
&& tar -xvf asio-1-12-2.tar.gz && cd asio-asio-1-12-2/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost \
&& wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
&& tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
&& ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests \
&& make install \
&& cd ../../ && rm -rf asio*
@@ -37,9 +37,9 @@ RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade c
RUN echo "*** Downloading RESTinio ***" \
&& mkdir restinio && cd restinio \
&& wget https://github.com/Stiffstream/restinio/archive/v.0.5.1.2.tar.gz \
&& ls -l && tar -xzf v.0.5.1.2.tar.gz \
&& cd restinio-v.0.5.1.2/dev \
&& wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
&& ls -l && tar -xzf a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
&& cd restinio-a7a10e419d9089c5b8ee63f5e3098c892f22fae4/dev \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
-DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
-DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
+1 -1
View File
@@ -3,5 +3,5 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
COPY . /root/opendht
RUN cd /root/opendht && mkdir build && cd build \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON .. \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON .. \
&& make -j8 && ./opendht_unit_tests && make install
+1 -1
View File
@@ -3,5 +3,5 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
COPY . /root/opendht
RUN cd /root/opendht && mkdir build && cd build \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_TESTS=ON .. \
&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_TESTS=ON .. \
&& make -j8 && ./opendht_unit_tests && make install
-6
View File
@@ -19,10 +19,4 @@
#pragma once
#include "opendht/dhtrunner.h"
#ifdef OPENDHT_PROXY_SERVER
#include "opendht/dht_proxy_server.h"
#endif
#include "opendht/default_types.h"
#ifdef OPENDHT_INDEXATION
#include "opendht/indexation/pht.h"
#endif
+3 -3
View File
@@ -76,9 +76,9 @@ struct OPENDHT_PUBLIC NodeStats {
struct OPENDHT_PUBLIC NodeInfo {
InfoHash id;
InfoHash node_id;
NodeStats ipv4;
NodeStats ipv6;
size_t ongoing_ops;
NodeStats ipv4 {};
NodeStats ipv6 {};
size_t ongoing_ops {0};
#ifdef OPENDHT_JSONCPP
/**
+5 -3
View File
@@ -114,6 +114,7 @@ struct OPENDHT_PUBLIC PublicKey
}
void pack(Blob& b) const;
int pack(uint8_t* out, size_t* out_len) const;
void unpack(const uint8_t* dat, size_t dat_size);
std::string toString() const;
@@ -127,7 +128,7 @@ struct OPENDHT_PUBLIC PublicKey
p.pack_bin_body((const char*)b.data(), b.size());
}
void msgpack_unpack(msgpack::object o);
void msgpack_unpack(const msgpack::object& o);
gnutls_digest_algorithm_t getPreferredDigest() const;
@@ -160,6 +161,7 @@ struct OPENDHT_PUBLIC PrivateKey
explicit operator bool() const { return key; }
PublicKey getPublicKey() const;
int serialize(uint8_t* out, size_t* out_len, const std::string& password = {}) const;
Blob serialize(const std::string& password = {}) const;
/**
@@ -224,7 +226,7 @@ public:
p.pack_bin_body((const char*)b.data(), b.size());
}
void msgpack_unpack(msgpack::object o);
void msgpack_unpack(const msgpack::object& o);
void revoke(const Certificate& crt, time_point t = time_point::min());
@@ -422,7 +424,7 @@ struct OPENDHT_PUBLIC Certificate {
p.pack_bin_body((const char*)b.data(), b.size());
}
void msgpack_unpack(msgpack::object o);
void msgpack_unpack(const msgpack::object& o);
explicit operator bool() const { return cert; }
PublicKey getPublicKey() const;
+2 -2
View File
@@ -189,7 +189,7 @@ public:
#endif
}
virtual void msgpack_unpack(msgpack::object o)
virtual void msgpack_unpack(const msgpack::object& o)
{
if (o.type != msgpack::type::ARRAY) throw msgpack::type_error();
if (o.via.array.size < 2) throw msgpack::type_error();
@@ -229,7 +229,7 @@ public:
pk.pack_bin_body((const char*)addr.get(), addr.getLength());
}
virtual void msgpack_unpack(msgpack::object o)
virtual void msgpack_unpack(const msgpack::object& o)
{
if (o.type == msgpack::type::BIN)
addr = {(sockaddr*)o.via.bin.ptr, (socklen_t)o.via.bin.size};
+5 -4
View File
@@ -53,9 +53,9 @@ public:
DhtProxyClient();
explicit DhtProxyClient(
std::shared_ptr<dht::crypto::Certificate> serverCA, dht::crypto::Identity clientIdentity,
std::shared_ptr<crypto::Certificate> serverCA, crypto::Identity clientIdentity,
std::function<void()> loopSignal, const std::string& serverHost,
const std::string& pushClientId = "", std::shared_ptr<dht::Logger> logger = {});
const std::string& pushClientId = "", std::shared_ptr<Logger> logger = {});
void setHeaderFields(http::Request& request);
@@ -303,7 +303,7 @@ private:
/**
* Send Listen with httpClient_
*/
void sendListen(const restinio::http_request_header_t header, const ValueCallback& cb,
void sendListen(const restinio::http_request_header_t& header, const ValueCallback& cb,
const Sp<OperationState>& opstate, Listener& listener, ListenMethod method = ListenMethod::LISTEN);
void handleResubscribe(const asio::error_code& ec, const InfoHash& key,
const size_t token, std::shared_ptr<OperationState> opstate);
@@ -327,6 +327,7 @@ private:
std::shared_ptr<dht::crypto::Certificate> serverCertificate_;
//std::pair<std::string, std::string> serverHostService_;
std::string pushClientId_;
std::string pushSessionId_;
mutable std::mutex lockCurrentProxyInfos_;
NodeStatus statusIpv4_ {NodeStatus::Disconnected};
@@ -404,7 +405,7 @@ private:
#endif // OPENDHT_PUSH_NOTIFICATIONS
Json::StreamWriterBuilder jsonBuilder_;
Json::CharReaderBuilder jsonReaderBuilder_;
std::unique_ptr<Json::CharReader> jsonReader_;
std::shared_ptr<dht::Logger> logger_;
std::shared_ptr<http::Request> buildRequest(const std::string& target = {});
+7 -13
View File
@@ -31,10 +31,7 @@
#include <restinio/all.hpp>
#include <restinio/tls.hpp>
#ifdef OPENDHT_JSONCPP
#include <json/json.h>
#endif
#include <memory>
#include <mutex>
@@ -54,8 +51,6 @@ class DhtRunner;
using RestRouter = restinio::router::express_router_t<>;
using RequestStatus = restinio::request_handling_status_t;
using ResponseByParts = restinio::chunked_output_t;
using ResponseByPartsBuilder = restinio::response_builder_t<ResponseByParts>;
/**
* Describes the REST API
@@ -87,15 +82,15 @@ public:
struct ServerStats {
/** Current number of listen operations */
size_t listenCount;
size_t listenCount {0};
/** Current number of permanent put operations */
size_t putCount;
size_t putCount {0};
/** Current number of push tokens with at least one listen operation */
size_t pushListenersCount;
size_t pushListenersCount {0};
/** Average requests per second */
double requestRate;
double requestRate {0};
/** Node Info **/
NodeInfo nodeInfo;
NodeInfo nodeInfo {};
std::string toString() const {
std::ostringstream ss;
@@ -113,7 +108,6 @@ public:
return ss.str();
}
#ifdef OPENDHT_JSONCPP
/**
* Build a json object from a NodeStats
*/
@@ -126,7 +120,6 @@ public:
result["nodeInfo"] = nodeInfo.toJson();
return result;
}
#endif
};
ServerStats stats() const { return stats_; }
@@ -288,7 +281,7 @@ private:
* @param key of the device
* @param json, the content to send
*/
void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid);
void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid, bool highPriority);
/**
* Send push notification with an expire timeout.
@@ -367,6 +360,7 @@ private:
#ifdef OPENDHT_PUSH_NOTIFICATIONS
struct Listener {
std::string clientId;
std::string sessionId;
std::future<size_t> internalToken;
std::unique_ptr<asio::steady_timer> expireTimer;
std::unique_ptr<asio::steady_timer> expireNotifyTimer;
+19 -20
View File
@@ -99,7 +99,7 @@ public:
asio::streambuf& input();
std::istream& data() { return istream_; }
std::string read_bytes(size_t bytes);
std::string read_bytes(size_t bytes = 0);
std::string read_until(const char delim);
void async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb);
@@ -108,11 +108,20 @@ public:
void async_read_until(const char* delim, BytesHandlerCb cb);
void async_read_until(char delim, BytesHandlerCb cb);
void async_read(size_t bytes, BytesHandlerCb cb);
void async_read_some(size_t bytes, BytesHandlerCb cb);
void timeout(const std::chrono::seconds timeout, HandlerCb cb = {});
void close();
private:
template<typename T>
T wrapCallabck(T cb) const {
return [t=shared_from_this(),cb=std::move(cb)](auto ...params) {
cb(params...);
};
}
unsigned int id_;
static std::atomic_uint ids_;
@@ -212,6 +221,7 @@ public:
using OnDataCb = std::function<void(const char* at, size_t length)>;
using OnStateChangeCb = std::function<void(State state, const Response& response)>;
using OnJsonCb = std::function<void(Json::Value value, unsigned int status_code)>;
using OnDoneCb = std::function<void(const Response& response)>;
// resolves implicitly
Request(asio::io_context& ctx, const std::string& url, const Json::Value& json, OnJsonCb jsoncb,
@@ -219,6 +229,7 @@ public:
Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<dht::Logger> logger = {});
Request(asio::io_context& ctx, const std::string& host, const std::string& service,
const bool ssl = false, std::shared_ptr<dht::Logger> logger = {});
Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger = {});
// user defined resolver
Request(asio::io_context& ctx, std::shared_ptr<Resolver> resolver, sa_family_t family = AF_UNSPEC);
@@ -230,7 +241,7 @@ public:
~Request();
unsigned int id() const;
inline unsigned int id() const { return id_; };
void set_connection(std::shared_ptr<Connection> connection);
std::shared_ptr<Connection> get_connection() const;
inline const Url& get_url() const {
@@ -258,6 +269,7 @@ public:
void add_on_status_callback(OnStatusCb cb);
void add_on_body_callback(OnDataCb cb);
void add_on_state_change_callback(OnStateChangeCb cb);
void add_on_done_callback(OnDoneCb cb);
void send();
@@ -270,19 +282,14 @@ private:
using OnCompleteCb = std::function<void()>;
struct Callbacks {
Callbacks(){}
OnStatusCb on_status;
OnDataCb on_header_field;
OnDataCb on_header_value;
OnDataCb on_body;
OnCompleteCb on_headers_complete;
OnCompleteCb on_message_complete;
OnStateChangeCb on_state_change;
};
void notify_state_change(const State state);
void notify_state_change(State state);
void build();
@@ -299,17 +306,11 @@ private:
void post();
void handle_request(const asio::error_code& ec);
void handle_response(const asio::error_code& ec, size_t bytes);
void handle_response_header(const asio::error_code& ec);
void handle_response_body(const asio::error_code& ec, size_t bytes);
/**
* Parse the request with http_parser.
* Return how many bytes were parsed.
* Note: we pass requerst.size()==0 to signal that EOF has been received.
*/
size_t parse_request(const std::string& request);
void onHeadersComplete();
void onBody(const char* at, size_t length);
void onComplete();
std::shared_ptr<dht::Logger> logger_;
@@ -318,7 +319,6 @@ private:
restinio::http_connection_header_t connection_type_ {restinio::http_connection_header_t::close};
std::string body_;
std::mutex cbs_mutex_;
Callbacks cbs_;
State state_;
@@ -336,7 +336,6 @@ private:
Response response_ {};
std::string request_;
std::atomic<bool> message_complete_ {false};
std::atomic<bool> finishing_ {false};
std::unique_ptr<http_parser> parser_;
std::unique_ptr<http_parser_settings> parser_s_;
+3 -3
View File
@@ -71,13 +71,13 @@ OPENDHT_PUBLIC void
printLog(std::ostream &s, char const *m, va_list args);
OPENDHT_PUBLIC
std::unique_ptr<Logger> getStdLogger();
std::shared_ptr<Logger> getStdLogger();
OPENDHT_PUBLIC
std::unique_ptr<Logger> getFileLogger(const std::string &path);
std::shared_ptr<Logger> getFileLogger(const std::string &path);
OPENDHT_PUBLIC
std::unique_ptr<Logger> getSyslogLogger(const char* name);
std::shared_ptr<Logger> getSyslogLogger(const char* name);
OPENDHT_PUBLIC void
enableLogging(dht::DhtRunner &dht);
+9 -14
View File
@@ -22,10 +22,13 @@
#include "def.h"
#include "sockaddr.h"
#include "infohash.h"
#include "log_enable.h"
#include <thread>
#include <mutex>
#include <condition_variable>
namespace asio {
class io_context;
}
namespace dht {
@@ -35,7 +38,7 @@ public:
static constexpr in_port_t DEFAULT_PORT = 8888;
using ServiceDiscoveredCallback = std::function<void(msgpack::object&&, SockAddr&&)>;
PeerDiscovery(in_port_t port = DEFAULT_PORT);
PeerDiscovery(in_port_t port = DEFAULT_PORT, std::shared_ptr<asio::io_context> ioContext = {}, std::shared_ptr<Logger> logger = {});
~PeerDiscovery();
/**
@@ -51,7 +54,7 @@ public:
}
/**
* startPublish - Keeping sending data until node is joinned or stop is called - msgpack
* 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);
@@ -79,20 +82,12 @@ public:
bool stopPublish(const std::string &type);
bool stopPublish(sa_family_t domain, const std::string &type);
/**
* Configure the sockopt to be able to listen multicast group
*/
static void joinMulticast(int sockfd, sa_family_t family);
/**
* Join the threads
*/
void join();
private:
class DomainPeerDiscovery;
std::unique_ptr<DomainPeerDiscovery> peerDiscovery4_;
std::unique_ptr<DomainPeerDiscovery> peerDiscovery6_;
std::shared_ptr<asio::io_context> ioContext_;
std::thread ioRunnner_;
};
}
+3 -2
View File
@@ -133,7 +133,7 @@ using Blob = std::vector<uint8_t>;
/**
* Provides backward compatibility with msgpack 1.0
*/
OPENDHT_PUBLIC Blob unpackBlob(msgpack::object& o);
OPENDHT_PUBLIC Blob unpackBlob(const msgpack::object& o);
template <typename Type>
Blob
@@ -153,6 +153,7 @@ unpackMsg(Blob b) {
msgpack::unpacked unpackMsg(Blob b);
msgpack::object* findMapValue(msgpack::object& map, const std::string& key);
msgpack::object* findMapValue(const msgpack::object& map, const char* key);
msgpack::object* findMapValue(const msgpack::object& map, const std::string& key);
} // namespace dht
+42 -17
View File
@@ -43,6 +43,19 @@
namespace dht {
static const std::string VALUE_KEY_ID("id");
static const std::string VALUE_KEY_DAT("dat");
static const std::string VALUE_KEY_PRIO("p");
static const std::string VALUE_KEY_SIGNATURE("sig");
static const std::string VALUE_KEY_SEQ("seq");
static const std::string VALUE_KEY_DATA("data");
static const std::string VALUE_KEY_OWNER("owner");
static const std::string VALUE_KEY_TYPE("type");
static const std::string VALUE_KEY_TO("to");
static const std::string VALUE_KEY_BODY("body");
static const std::string VALUE_KEY_USERTYPE("utype");
struct Value;
struct Query;
@@ -390,7 +403,7 @@ struct OPENDHT_PUBLIC Value
* Build a value from a json object
* @param json
*/
Value(Json::Value& json);
Value(const Json::Value& json);
#endif
template <typename Type>
@@ -408,7 +421,9 @@ struct OPENDHT_PUBLIC Value
Value(Value&& o) noexcept
: id(o.id), owner(std::move(o.owner)), recipient(o.recipient),
type(o.type), data(std::move(o.data)), user_type(std::move(o.user_type)), seq(o.seq), signature(std::move(o.signature)), cypher(std::move(o.cypher)) {}
type(o.type), data(std::move(o.data)), user_type(std::move(o.user_type)), seq(o.seq)
, signature(std::move(o.signature)), cypher(std::move(o.cypher))
, priority(o.priority) {}
template <typename Type>
Value(const Type& vs)
@@ -485,17 +500,17 @@ struct OPENDHT_PUBLIC Value
bool has_owner = owner && *owner;
pk.pack_map((user_type.empty()?0:1) + (has_owner?(recipient ? 5 : 4):2));
if (has_owner) { // isSigned
pk.pack(std::string("seq")); pk.pack(seq);
pk.pack(std::string("owner")); owner->msgpack_pack(pk);
pk.pack(VALUE_KEY_SEQ); pk.pack(seq);
pk.pack(VALUE_KEY_OWNER); owner->msgpack_pack(pk);
if (recipient) {
pk.pack(std::string("to")); pk.pack(recipient);
pk.pack(VALUE_KEY_TO); pk.pack(recipient);
}
}
pk.pack(std::string("type")); pk.pack(type);
pk.pack(std::string("data")); pk.pack_bin(data.size());
pk.pack_bin_body((const char*)data.data(), data.size());
pk.pack(VALUE_KEY_TYPE); pk.pack(type);
pk.pack(VALUE_KEY_DATA); pk.pack_bin(data.size());
pk.pack_bin_body((const char*)data.data(), data.size());
if (not user_type.empty()) {
pk.pack(std::string("utype")); pk.pack(user_type);
pk.pack(VALUE_KEY_USERTYPE); pk.pack(user_type);
}
}
@@ -507,10 +522,10 @@ struct OPENDHT_PUBLIC Value
pk.pack_bin_body((const char*)cypher.data(), cypher.size());
} else {
pk.pack_map(isSigned() ? 2 : 1);
pk.pack(std::string("body")); msgpack_pack_to_sign(pk);
pk.pack(VALUE_KEY_BODY); msgpack_pack_to_sign(pk);
if (isSigned()) {
pk.pack(std::string("sig")); pk.pack_bin(signature.size());
pk.pack_bin_body((const char*)signature.data(), signature.size());
pk.pack(VALUE_KEY_SIGNATURE); pk.pack_bin(signature.size());
pk.pack_bin_body((const char*)signature.data(), signature.size());
}
}
}
@@ -518,9 +533,12 @@ struct OPENDHT_PUBLIC Value
template <typename Packer>
void msgpack_pack(Packer& pk) const
{
pk.pack_map(2);
pk.pack(std::string("id")); pk.pack(id);
pk.pack(std::string("dat")); msgpack_pack_to_encrypt(pk);
pk.pack_map(2 + (priority?1:0));
pk.pack(VALUE_KEY_ID); pk.pack(id);
pk.pack(VALUE_KEY_DAT); msgpack_pack_to_encrypt(pk);
if (priority) {
pk.pack(VALUE_KEY_PRIO); pk.pack(priority);
}
}
template <typename Packer>
@@ -551,7 +569,7 @@ struct OPENDHT_PUBLIC Value
}
}
void msgpack_unpack(msgpack::object o);
void msgpack_unpack(const msgpack::object& o);
void msgpack_unpack_body(const msgpack::object& o);
Blob getPacked() const {
msgpack::sbuffer buffer;
@@ -602,6 +620,13 @@ struct OPENDHT_PUBLIC Value
*/
Blob cypher {};
/**
* Priority of this value (used for push notifications)
* 0 = high priority
* 1 = normal priority
*/
unsigned priority {0};
private:
friend class SecureDht;
/* Cache for crypto ops */
@@ -658,7 +683,7 @@ struct OPENDHT_PUBLIC FieldValue
}
}
void msgpack_unpack(msgpack::object msg) {
void msgpack_unpack(const msgpack::object& msg) {
hashValue = {};
blobValue.clear();
+5 -2
View File
@@ -27,7 +27,6 @@ libopendht_la_SOURCES = \
dhtrunner.cpp \
default_types.cpp \
log.cpp \
peer_discovery.cpp \
network_utils.cpp \
thread_pool.cpp
@@ -56,7 +55,6 @@ nobase_include_HEADERS = \
../include/opendht/default_types.h \
../include/opendht/log.h \
../include/opendht/log_enable.h \
../include/opendht/peer_discovery.h \
../include/opendht/network_utils.h \
../include/opendht/rng.h \
../include/opendht/thread_pool.h
@@ -77,6 +75,11 @@ libopendht_la_SOURCES += http.cpp
nobase_include_HEADERS += ../include/opendht/proxy.h ../include/opendht/http.h
endif
if ENABLE_PEER_DISCOVERY
libopendht_la_SOURCES += peer_discovery.cpp
nobase_include_HEADERS += ../include/opendht/peer_discovery.h
endif
if ENABLE_INDEXATION
libopendht_la_SOURCES += indexation/pht.cpp
nobase_include_HEADERS += ../include/opendht/indexation/pht.h
+2 -2
View File
@@ -131,7 +131,7 @@ base64_encode(const std::vector<unsigned char>& str)
return base64_encode(str.cbegin(), str.cend());
}
std::string
std::vector<unsigned char>
base64_decode(const std::string& str)
{
size_t output_length = str.length() / 4 * 3 + 2;
@@ -139,5 +139,5 @@ base64_decode(const std::string& str)
output.resize(output_length);
base64_decode(str.data(), str.size(), output.data(), &output_length);
output.resize(output_length);
return std::string(output.begin(), output.end());
return output;
}
+1 -1
View File
@@ -33,4 +33,4 @@ std::string base64_encode(const std::vector<unsigned char>& str);
* @param str the input buffer
* @return a base64-decoded buffer
*/
std::string base64_decode(const std::string& str);
std::vector<unsigned char> base64_decode(const std::string& str);
+22
View File
@@ -1,3 +1,25 @@
/*
* Copyright (C) 2014-2019 Savoir-faire Linux Inc.
* Author(s) : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "callbacks.h"
namespace dht {
+21 -7
View File
@@ -348,9 +348,7 @@ PrivateKey::serialize(const std::string& password) const
size_t buf_sz = 8192;
Blob buffer;
buffer.resize(buf_sz);
int err = password.empty()
? gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, nullptr, GNUTLS_PKCS_PLAIN, buffer.data(), &buf_sz)
: gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, password.c_str(), GNUTLS_PKCS_PBES2_AES_256, buffer.data(), &buf_sz);
auto err = serialize(buffer.data(), &buf_sz, password);
if (err != GNUTLS_E_SUCCESS) {
std::cerr << "Could not export private key - " << gnutls_strerror(err) << std::endl;
return {};
@@ -359,6 +357,16 @@ PrivateKey::serialize(const std::string& password) const
return buffer;
}
int
PrivateKey::serialize(uint8_t* out, size_t* out_len, const std::string& password) const
{
if (!x509_key)
return -1;
return password.empty()
? gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, nullptr, GNUTLS_PKCS_PLAIN, out, out_len)
: gnutls_x509_privkey_export_pkcs8(x509_key, GNUTLS_X509_FMT_PEM, password.c_str(), GNUTLS_PKCS_PBES2_AES_256, out, out_len);
}
PublicKey
PrivateKey::getPublicKey() const
{
@@ -404,12 +412,18 @@ PublicKey::pack(Blob& b) const
throw CryptoException(std::string("Could not export public key: null key"));
std::vector<uint8_t> tmp(2048);
size_t sz = tmp.size();
if (int err = gnutls_pubkey_export(pk, GNUTLS_X509_FMT_DER, tmp.data(), &sz))
if (int err = pack(tmp.data(), &sz))
throw CryptoException(std::string("Could not export public key: ") + gnutls_strerror(err));
tmp.resize(sz);
b.insert(b.end(), tmp.begin(), tmp.end());
}
int
PublicKey::pack(uint8_t* out, size_t* out_len) const
{
return gnutls_pubkey_export(pk, GNUTLS_X509_FMT_DER, out, out_len);
}
void
PublicKey::unpack(const uint8_t* data, size_t data_size)
{
@@ -439,7 +453,7 @@ PublicKey::toString() const
}
void
PublicKey::msgpack_unpack(msgpack::object o)
PublicKey::msgpack_unpack(const msgpack::object& o)
{
if (o.type == msgpack::type::BIN)
unpack((const uint8_t*)o.via.bin.ptr, o.via.bin.size);
@@ -784,7 +798,7 @@ Certificate::unpack(const uint8_t* dat, size_t dat_size)
}
void
Certificate::msgpack_unpack(msgpack::object o)
Certificate::msgpack_unpack(const msgpack::object& o)
{
if (o.type == msgpack::type::BIN)
unpack((const uint8_t*)o.via.bin.ptr, o.via.bin.size);
@@ -1244,7 +1258,7 @@ RevocationList::unpack(const uint8_t* dat, size_t dat_size)
}
void
RevocationList::msgpack_unpack(msgpack::object o)
RevocationList::msgpack_unpack(const msgpack::object& o)
{
try {
if (o.type == msgpack::type::BIN)
+2 -1
View File
@@ -76,8 +76,9 @@ Dht::shutdown(ShutdownCallback cb)
for (auto& str : store)
*remaining += maintainStorage(str, true, str_donecb);
DHT_LOG.w("shuting down node: after storage, %u ops", *remaining);
if (!*remaining) {
DHT_LOG.w("shuting down node: %u ops remaining", *remaining);
if (cb) cb();
}
}
+166 -137
View File
@@ -24,6 +24,7 @@
#include "utils.h"
#include <http_parser.h>
#include <deque>
namespace dht {
@@ -69,22 +70,55 @@ struct DhtProxyClient::ProxySearch {
std::map<Value::Id, PermanentPut> puts {};
};
struct LineSplit {
void append(const char* d, size_t l) {
buf_.insert(buf_.end(), d, d+l);
}
bool getLine(char c) {
auto it = buf_.begin();
while (it != buf_.end()) {
if (*it == c) {
line_.clear();
line_.insert(line_.end(), buf_.begin(), ++it);
buf_.erase(buf_.begin(), it);
return true;
}
it++;
}
return false;
}
const std::string& line() const { return line_; }
private:
std::deque<char> buf_ {};
std::string line_ {};
};
std::string
getRandomSessionId(size_t length = 8) {
static constexpr const char chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~";
std::string str(length, 0);
crypto::random_device rdev;
std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 1);
std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } );
return str;
}
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& pushClientId, std::shared_ptr<dht::Logger> logger)
:
clientIdentity_(clientIdentity), serverCertificate_(serverCA),
pushClientId_(pushClientId), loopSignal_(signal), logger_(logger)
: proxyUrl_(serverHost)
, clientIdentity_(clientIdentity), serverCertificate_(serverCA)
, pushClientId_(pushClientId), pushSessionId_(getRandomSessionId())
, loopSignal_(signal)
, jsonReader_(Json::CharReaderBuilder{}.newCharReader())
, logger_(logger)
{
// build http client
proxyUrl_ = serverHost;
jsonBuilder_["commentStyle"] = "None";
jsonBuilder_["indentation"] = "";
if (logger_){
if (logger_) {
if (serverCertificate_)
logger_->d("[proxy:client] using ca certificate for ssl:\n%s",
serverCertificate_->toString(false/*chain*/).c_str());
@@ -152,21 +186,22 @@ DhtProxyClient::~DhtProxyClient()
void
DhtProxyClient::stop()
{
isDestroying_ = true;
resolver_.reset();
cancelAllListeners();
if (infoState_)
infoState_->cancel = true;
{
std::lock_guard<std::mutex> lock(requestLock_);
for (auto& request : requests_)
request.second->cancel();
if (not isDestroying_.exchange(true)) {
resolver_.reset();
cancelAllListeners();
if (infoState_)
infoState_->cancel = true;
{
std::lock_guard<std::mutex> lock(requestLock_);
for (auto& request : requests_)
request.second->cancel();
}
if (not httpContext_.stopped())
httpContext_.stop();
if (httpClientThread_.joinable())
httpClientThread_.join();
requests_.clear();
}
if (not httpContext_.stopped())
httpContext_.stop();
if (httpClientThread_.joinable())
httpClientThread_.join();
requests_.clear();
}
std::vector<Sp<Value>>
@@ -271,28 +306,31 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
if (logger_)
logger_->d("[proxy:client] [get] [search %s]", key.to_c_str());
if (isDestroying_) {
if (donecb) donecb(false, {});
return;
}
try {
auto request = buildRequest("/" + key.toString());
auto reqid = request->id();
request->set_connection_type(restinio::http_connection_header_t::keep_alive);
//request->set_connection_type(restinio::http_connection_header_t::keep_alive);
request->set_method(restinio::http_method_get());
setHeaderFields(*request);
auto opstate = std::make_shared<OperationState>();
Value::Filter filter = w.empty() ? f : f.chain(w.getFilter());
request->add_on_body_callback([this, key, opstate, filter, cb](const char* at, size_t length){
auto rxBuf = std::make_shared<LineSplit>();
request->add_on_body_callback([this, key, opstate, filter, rxBuf, cb](const char* at, size_t length){
try {
auto body = std::string(at, length);
auto& b = *rxBuf;
b.append(at, length);
// one value per body line
std::string data_line;
std::stringstream body_stream(body);
while (std::getline(body_stream, data_line, '\n') and !(opstate->stop.load())){
while (b.getLine('\n') and !opstate->stop){
std::string err;
Json::Value json;
auto* char_data = static_cast<const char*>(&data_line[0]);
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){
const auto& line = b.line();
if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){
opstate->ok.store(false);
return;
}
@@ -315,30 +353,27 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
opstate->ok.store(false);
}
});
request->add_on_state_change_callback([this, reqid, opstate, donecb, key]
(http::Request::State state, const http::Response& response){
if (state == http::Request::State::DONE){
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code);
opstate->ok.store(false);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (donecb) {
{
std::lock_guard<std::mutex> lock(lockCallbacks_);
callbacks_.emplace_back([donecb, opstate](){
donecb(opstate->ok, {});
opstate->stop.store(true);
});
}
loopSignal_();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
request->add_on_done_callback([this, reqid, opstate, donecb, key] (const http::Response& response){
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [get %s] failed with code=%i", key.to_c_str(), response.status_code);
opstate->ok.store(false);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (donecb) {
{
std::lock_guard<std::mutex> lock(lockCallbacks_);
callbacks_.emplace_back([donecb, opstate](){
donecb(opstate->ok, {});
opstate->stop.store(true);
});
}
loopSignal_();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
});
{
@@ -356,9 +391,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
void
DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_point created, bool permanent)
{
if (not val){
if (cb)
cb(false, {});
if (not val or isDestroying_) {
if (cb) cb(false, {});
return;
}
if (logger_)
@@ -465,22 +499,19 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
}
}
request->set_body(Json::writeString(jsonBuilder_, json));
request->add_on_state_change_callback([this, reqid, cb]
(http::Request::State state, const http::Response& response){
if (state == http::Request::State::DONE){
bool ok = response.status_code == 200;
if (not ok) {
if (logger_)
logger_->e("[proxy:client] [status] failed with code=%i", response.status_code);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (cb)
cb(ok);
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
request->add_on_done_callback([this, reqid, cb] (const http::Response& response){
bool ok = response.status_code == 200;
if (not ok) {
if (logger_)
logger_->e("[proxy:client] [status] failed with code=%i", response.status_code);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (cb)
cb(ok);
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
});
{
@@ -580,33 +611,29 @@ DhtProxyClient::queryProxyInfo(std::shared_ptr<InfoState> infoState, sa_family_t
auto reqid = request->id();
request->set_method(restinio::http_method_get());
setHeaderFields(*request);
request->add_on_state_change_callback([this, reqid, family, infoState]
(http::Request::State state, const http::Response& response){
if (state == http::Request::State::DONE) {
if (infoState->cancel.load())
request->add_on_done_callback([this, reqid, family, infoState] (const http::Response& response){
if (infoState->cancel.load())
return;
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [status] ipv%i failed with code=%i",
family == AF_INET ? 4 : 6, response.status_code);
// pass along the failures
if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0))
onProxyInfos(Json::Value{}, family);
} else {
std::string err;
Json::Value proxyInfos;
if (!jsonReader_->parse(response.body.data(), response.body.data() + response.body.size(), &proxyInfos, &err)){
onProxyInfos(Json::Value{}, family);
return;
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [status] ipv%i failed with code=%i",
family == AF_INET ? 4 : 6, response.status_code);
// pass along the failures
if ((family == AF_INET and infoState->ipv4 == 0) or (family == AF_INET6 and infoState->ipv6 == 0))
onProxyInfos(Json::Value{}, family);
} else {
std::string err;
Json::Value proxyInfos;
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &proxyInfos, &err)){
onProxyInfos(Json::Value{}, family);
return;
}
if (not infoState->cancel)
onProxyInfos(proxyInfos, family);
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
if (not infoState->cancel)
onProxyInfos(proxyInfos, family);
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
});
@@ -703,6 +730,8 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
{
if (logger_)
logger_->d("[proxy:client] [listen] [search %s]", key.to_c_str());
if (isDestroying_)
return 0;
std::lock_guard<std::mutex> lock(searchLock_);
auto& search = searches_[key];
@@ -868,20 +897,17 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash&
body["key"] = deviceKey_;
body["client_id"] = pushClientId_;
request->set_body(Json::writeString(jsonBuilder_, body));
request->add_on_state_change_callback([this, reqid, key]
(http::Request::State state, const http::Response& response){
if (state == http::Request::State::DONE){
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i",
key.to_c_str(), response.status_code);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
request->add_on_done_callback([this, reqid, key] (const http::Response& response){
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [unsubscribe %s] failed with code=%i",
key.to_c_str(), response.status_code);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
});
{
@@ -914,7 +940,7 @@ DhtProxyClient::handleExpireListener(const asio::error_code &ec, const InfoHash&
}
void
DhtProxyClient::sendListen(const restinio::http_request_header_t header,
DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
const ValueCallback& cb,
const Sp<OperationState>& opstate,
Listener& listener, ListenMethod method)
@@ -935,19 +961,18 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
body = fillBody(method == ListenMethod::RESUBSCRIBE);
request->set_body(body);
#endif
request->add_on_body_callback([this, reqid, opstate, cb](const char* at, size_t length){
auto rxBuf = std::make_shared<LineSplit>();
request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length){
try {
auto body = std::string(at, length);
auto& b = *rxBuf;
b.append(at, length);
// one value per body line
std::string data_line;
std::stringstream body_stream(body);
while (std::getline(body_stream, data_line, '\n') and !opstate->stop){
while (b.getLine('\n') and !opstate->stop){
std::string err;
Json::Value json;
auto* char_data = static_cast<const char*>(&data_line[0]);
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(char_data, char_data + data_line.size(), &json, &err)){
const auto& line = b.line();
if (!jsonReader_->parse(line.data(), line.data() + line.size(), &json, &err)){
opstate->ok.store(false);
return;
}
@@ -974,21 +999,18 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header,
opstate->ok.store(false);
}
});
request->add_on_state_change_callback([this, opstate, reqid]
(http::Request::State state, const http::Response& response){
if (state == http::Request::State::DONE){
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [listen] send request #%i failed with code=%i",
reqid, response.status_code);
opstate->ok.store(false);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
request->add_on_done_callback([this, opstate, reqid] (const http::Response& response) {
if (response.status_code != 200) {
if (logger_)
logger_->e("[proxy:client] [listen] send request #%i failed with code=%i",
reqid, response.status_code);
opstate->ok.store(false);
if (not response.aborted and response.status_code == 0)
opFailed();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
requests_.erase(reqid);
}
});
{
@@ -1107,6 +1129,12 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
statusIpv6_ = NodeStatus::Connected;
}
try {
auto sessionId = notification.find("s");
if (sessionId != notification.end() and sessionId->second != pushSessionId_) {
if (logger_)
logger_->d("[proxy:client] [push] ignoring push for other session");
return;
}
std::lock_guard<std::mutex> lock(searchLock_);
auto timeout = notification.find("timeout");
if (timeout != notification.cend()) {
@@ -1151,7 +1179,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
} else {
std::stringstream ss(expired->second);
std::vector<Value::Id> ids;
while(ss.good()){
while(ss.good()) {
std::string substr;
getline(ss, substr, ',');
ids.emplace_back(std::stoull(substr));
@@ -1226,6 +1254,7 @@ DhtProxyClient::getPushRequest(Json::Value& body) const
{
body["key"] = deviceKey_;
body["client_id"] = pushClientId_;
body["session_id"] = pushSessionId_;
#ifdef __ANDROID__
body["platform"] = "android";
#endif
+31 -20
View File
@@ -18,6 +18,10 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "dht_proxy_server.h"
#include "default_types.h"
@@ -68,6 +72,9 @@ constexpr char RESP_MSG_NO_TOKEN[] = "{\"err\":\"No token\"}";
constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2};
using ResponseByParts = restinio::chunked_output_t;
using ResponseByPartsBuilder = restinio::response_builder_t<ResponseByParts>;
class opendht_logger_t
{
public:
@@ -386,7 +393,6 @@ HttpResponse DhtProxyServer::initHttpResponse(HttpResponse response) const
response.append_header("Server", "RESTinio");
response.append_header(restinio::http_field::content_type, "application/json");
response.append_header(restinio::http_field::access_control_allow_origin, "*");
response.connection_keep_alive();
return response;
}
@@ -616,15 +622,16 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
}
try {
std::string err;
Json::Value root;
Json::Value r;
auto* char_data = reinterpret_cast<const char*>(request->body().data());
auto reader = std::unique_ptr<Json::CharReader>(jsonReaderBuilder_.newCharReader());
if (!reader->parse(char_data, char_data + request->body().size(), &root, &err)){
if (!reader->parse(char_data, char_data + request->body().size(), &r, &err)){
auto response = initHttpResponse(
request->create_response(restinio::status_bad_request()));
response.set_body(RESP_MSG_JSON_INCORRECT);
return response.done();
}
const Json::Value& root(r); // parse using const Json so [] never creates element
auto pushToken = root["key"].asString();
if (pushToken.empty()){
auto response = initHttpResponse(
@@ -632,12 +639,12 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
response.set_body(RESP_MSG_NO_TOKEN);
return response.done();
}
auto platform = root["platform"].asString();
auto isAndroid = platform == "android";
auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string();
auto isAndroid = root["platform"].asString() == "android";
auto clientId = root["client_id"].asString();
auto sessionId = root["session_id"].asString();
if (logger_)
logger_->d("[proxy:server] [subscribe %s] [client %s]", infoHash.toString().c_str(), clientId.c_str());
logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
// ================ Search for existing listener ===================
// start the timer
auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
@@ -647,11 +654,11 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first;
auto pushListeners = pushListener->second.listeners.emplace(infoHash, std::vector<Listener>{}).first;
for (auto &listener: pushListeners->second){
if (logger_)
logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
for (auto &listener: pushListeners->second) {
// Found -> Resubscribe
if (listener.clientId == clientId){
if (listener.clientId == clientId) {
if (logger_)
logger_->d("[proxy:server] [subscribe] found [client %s]", listener.clientId.c_str());
// Reset timers
listener.expireTimer->expires_at(timeout);
listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
@@ -686,12 +693,13 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
// Add listen on dht
listener.internalToken = dht_->listen(infoHash,
[this, infoHash, pushToken, isAndroid, clientId]
[this, infoHash, pushToken, isAndroid, clientId, sessionId]
(const std::vector<std::shared_ptr<Value>>& values, bool expired){
// Build message content
Json::Value json;
json["key"] = infoHash.toString();
json["to"] = clientId;
json["s"] = sessionId;
if (expired and values.size() < 2){
std::stringstream ss;
for(size_t i = 0; i < values.size(); ++i){
@@ -700,12 +708,15 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
}
json["exp"] = ss.str();
}
sendPushNotification(pushToken, std::move(json), isAndroid);
auto maxPrio = 1000u;
for (const auto& v : values)
maxPrio = std::min(maxPrio, v->priority);
sendPushNotification(pushToken, std::move(json), isAndroid, !expired and maxPrio == 0);
return true;
}
);
// Launch timers
auto &ctx = io_context();
auto& ctx = io_context();
// expire notify
if (!listener.expireNotifyTimer)
listener.expireNotifyTimer = std::make_unique<asio::steady_timer>(ctx, timeout - proxy::OP_MARGIN);
@@ -714,8 +725,9 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
Json::Value json;
json["timeout"] = infoHash.toString();
json["to"] = clientId;
json["s"] = sessionId;
listener.expireNotifyTimer->async_wait(std::bind(&DhtProxyServer::handleNotifyPushListenExpire, this,
std::placeholders::_1, pushToken, json, isAndroid));
std::placeholders::_1, pushToken, std::move(json), isAndroid));
// cancel push listen
if (!listener.expireTimer)
listener.expireTimer = std::make_unique<asio::steady_timer>(ctx, timeout);
@@ -796,7 +808,7 @@ DhtProxyServer::handleNotifyPushListenExpire(const asio::error_code &ec, const s
}
if (logger_)
logger_->d("[proxy:server] [subscribe] sending put refresh to %s token", pushToken.c_str());
sendPushNotification(pushToken, std::move(json), isAndroid);
sendPushNotification(pushToken, std::move(json), isAndroid, false);
}
void
@@ -838,7 +850,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code &ec, const std::st
}
void
DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid)
DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid, bool highPriority)
{
if (pushServer_.empty())
return;
@@ -862,7 +874,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
notification["tokens"] = std::move(tokens);
notification["platform"] = isAndroid ? 2 : 1;
notification["data"] = std::move(json);
notification["priority"] = "high";
notification["priority"] = highPriority ? "high" : "normal";
notification["time_to_live"] = 600;
Json::Value notifications(Json::arrayValue);
@@ -957,7 +969,7 @@ DhtProxyServer::put(restinio::request_handle_t request,
if (logger_)
logger_->d("[proxy:server] [put %s] %s %s", infoHash.toString().c_str(),
value->toString().c_str(), (permanent ? "permanent" : ""));
if (permanent){
if (permanent) {
std::string pushToken, clientId, platform;
auto& pVal = root["permanent"];
if (pVal.isObject()){
@@ -1003,7 +1015,6 @@ DhtProxyServer::put(restinio::request_handle_t request,
if (pput.expireNotifyTimer)
pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN);
}
lock.unlock();
}
dht_->put(infoHash, value, [this, request, value](bool ok){
if (ok){
+24 -11
View File
@@ -18,11 +18,16 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "dhtrunner.h"
#include "securedht.h"
#include "peer_discovery.h"
#include "network_utils.h"
#ifdef OPENDHT_PEER_DISCOVERY
#include "peer_discovery.h"
#endif
#ifdef OPENDHT_PROXY_CLIENT
#include "dht_proxy_client.h"
#endif
@@ -124,7 +129,8 @@ DhtRunner::run(const Config& config, Context&& context)
{
std::lock_guard<std::mutex> lck(sock_mtx);
if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
std::cerr << "Dropping packet: queue is full!" << std::endl;
if (logger_)
logger_->e("Dropping packet: queue is full!");
rcv.pop();
}
rcv.emplace(std::move(pkt));
@@ -184,11 +190,16 @@ DhtRunner::run(const Config& config, Context&& context)
});
if (config.peer_discovery or config.peer_publish) {
#ifdef OPENDHT_PEER_DISCOVERY
peerDiscovery_ = context.peerDiscovery ?
std::move(context.peerDiscovery) :
std::make_shared<PeerDiscovery>();
#else
std::cerr << "Peer discovery requested but OpenDHT built without peer discovery support." << std::endl;
#endif
}
#ifdef OPENDHT_PEER_DISCOVERY
auto netId = config.dht_config.node_config.network;
if (config.peer_discovery) {
peerDiscovery_->startDiscovery<NodeInsertionPack>(PEER_DISCOVERY_DHT_SERVICE, [this, netId](NodeInsertionPack&& v, SockAddr&& addr){
@@ -217,6 +228,7 @@ DhtRunner::run(const Config& config, Context&& context)
peerDiscovery_->startPublish(AF_INET6, PEER_DISCOVERY_DHT_SERVICE, sbuf_node);
}
}
#endif
}
void
@@ -292,8 +304,10 @@ DhtRunner::join()
return;
cv.notify_all();
bootstrap_cv.notify_all();
#ifdef OPENDHT_PEER_DISCOVERY
if (peerDiscovery_)
peerDiscovery_->stop();
#endif
if (dht_)
if (auto sock = dht_->getSocket())
sock->stop();
@@ -307,9 +321,6 @@ DhtRunner::join()
if (bootstrap_thread.joinable())
bootstrap_thread.join();
if (peerDiscovery_)
peerDiscovery_->join();
{
std::lock_guard<std::mutex> lck(storage_mtx);
pending_ops = decltype(pending_ops)();
@@ -465,12 +476,14 @@ DhtRunner::getNodesStats(sa_family_t af) const
NodeInfo
DhtRunner::getNodeInfo() const {
std::lock_guard<std::mutex> lck(dht_mtx);
NodeInfo info;
info.id = getId();
info.node_id = getNodeId();
NodeInfo info {};
if (auto dht = activeDht()) {
info.id = dht->getId();
info.node_id = dht->getNodeId();
info.ipv4 = dht->getNodesStats(AF_INET);
info.ipv6 = dht->getNodesStats(AF_INET6);
}
info.ongoing_ops = ongoing_ops;
info.ipv4 = dht_->getNodesStats(AF_INET);
info.ipv6 = dht_->getNodesStats(AF_INET6);
return info;
}
+168 -310
View File
@@ -29,15 +29,8 @@
namespace dht {
namespace http {
constexpr char HTTP_HEADER_CONNECTION[] = "Connection";
constexpr char HTTP_HEADER_CONNECTION_KEEP_ALIVE[] = "keep-alive";
constexpr char HTTP_HEADER_CONNECTION_CLOSE[] = "close";
constexpr char HTTP_HEADER_CONTENT_LENGTH[] = "Content-Length";
constexpr char HTTP_HEADER_CONTENT_TYPE_JSON[] = "application/json";
constexpr char HTTP_HEADER_TRANSFER_ENCODING[] = "Transfer-Encoding";
constexpr char HTTP_HEADER_TRANSFER_ENCODING_CHUNKED[] = "chunked";
constexpr char HTTP_HEADER_DELIM[] = "\r\n\r\n";
constexpr char BODY_VALUE_DELIM = '\n';
Url::Url(const std::string& url): url(url)
{
@@ -130,8 +123,7 @@ Connection::Connection(asio::io_context& ctx, std::shared_ptr<dht::crypto::Certi
ssl_socket_ = std::make_unique<ssl_socket_t>(ctx_, ssl_ctx_);
}
Connection::~Connection()
{
Connection::~Connection() {
close();
}
@@ -200,6 +192,8 @@ Connection::input()
std::string
Connection::read_bytes(size_t bytes)
{
if (bytes == 0)
bytes = read_buf_.in_avail();
std::string content;
content.resize(bytes);
auto rb = istream_.readsome(&content[0], bytes);
@@ -219,9 +213,9 @@ void
Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
{
if (ssl_socket_)
asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), cb);
asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(cb)));
else if (socket_)
asio::async_connect(*socket_, std::move(endpoints), cb);
asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
else if (cb)
cb(asio::error::operation_aborted, {});
}
@@ -261,8 +255,8 @@ void
Connection::async_write(BytesHandlerCb cb)
{
if (!is_open()) return;
if (ssl_socket_) asio::async_write(*ssl_socket_, write_buf_, cb);
else if (socket_) asio::async_write(*socket_, write_buf_, cb);
if (ssl_socket_) asio::async_write(*ssl_socket_, write_buf_, wrapCallabck(std::move(cb)));
else if (socket_) asio::async_write(*socket_, write_buf_, wrapCallabck(std::move(cb)));
else if (cb) cb(asio::error::operation_aborted, 0);
}
@@ -270,8 +264,8 @@ void
Connection::async_read_until(const char* delim, BytesHandlerCb cb)
{
if (!is_open()) return;
if (ssl_socket_) asio::async_read_until(*ssl_socket_, read_buf_, delim, cb);
else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, cb);
if (ssl_socket_) asio::async_read_until(*ssl_socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
else if (cb) cb(asio::error::operation_aborted, 0);
}
@@ -279,8 +273,8 @@ void
Connection::async_read_until(char delim, BytesHandlerCb cb)
{
if (!is_open()) return;
if (ssl_socket_) asio::async_read_until(*ssl_socket_, read_buf_, delim, cb);
else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, cb);
if (ssl_socket_) asio::async_read_until(*ssl_socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
else if (socket_) asio::async_read_until(*socket_, read_buf_, delim, wrapCallabck(std::move(cb)));
else if (cb) cb(asio::error::operation_aborted, 0);
}
@@ -288,11 +282,27 @@ void
Connection::async_read(size_t bytes, BytesHandlerCb cb)
{
if (!is_open()) return;
if (ssl_socket_) asio::async_read(*ssl_socket_, read_buf_, asio::transfer_exactly(bytes), cb);
else if (socket_) asio::async_read(*socket_, read_buf_, asio::transfer_exactly(bytes), cb);
if (ssl_socket_) asio::async_read(*ssl_socket_, read_buf_, asio::transfer_exactly(bytes), wrapCallabck(std::move(cb)));
else if (socket_) asio::async_read(*socket_, read_buf_, asio::transfer_exactly(bytes), wrapCallabck(std::move(cb)));
else if (cb) cb(asio::error::operation_aborted, 0);
}
void
Connection::async_read_some(size_t bytes, BytesHandlerCb cb)
{
if (!is_open()) {
if (cb) cb(asio::error::operation_aborted, 0);
return;
}
auto buf = read_buf_.prepare(bytes);
auto onEnd = [this_=shared_from_this(), cb=std::move(cb)](const asio::error_code& ec, size_t t){
this_->read_buf_.commit(t);
cb(ec, t);
};
if (ssl_socket_) ssl_socket_->async_read_some(buf, onEnd);
else socket_->async_read_some(buf, onEnd);
}
void
Connection::timeout(const std::chrono::seconds timeout, HandlerCb cb)
{
@@ -438,18 +448,19 @@ Request::Request(asio::io_context& ctx, const std::string& url, const Json::Valu
set_header_field(restinio::http_field_t::content_type, HTTP_HEADER_CONTENT_TYPE_JSON);
set_header_field(restinio::http_field_t::accept, HTTP_HEADER_CONTENT_TYPE_JSON);
Json::StreamWriterBuilder wbuilder;
set_method(restinio::http_method_post());
set_body(Json::writeString(wbuilder, json));
add_on_state_change_callback([this, jsoncb](State state, const Response& response){
if (state != Request::State::DONE)
return;
add_on_done_callback([this, jsoncb](const Response& response){
Json::Value json;
std::string err;
Json::CharReaderBuilder rbuilder;
auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &json, &err) and logger_)
logger_->e("[http:request:%i] can't parse response to json", id_, err.c_str());
if (response.status_code != 0) {
std::string err;
Json::CharReaderBuilder rbuilder;
auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
if (!reader->parse(response.body.data(), response.body.data() + response.body.size(), &json, &err) and logger_)
logger_->e("[http:request:%i] can't parse response to json", id_, err.c_str());
}
if (jsoncb)
jsoncb(json, response.status_code);
jsoncb(std::move(json), response.status_code);
});
}
@@ -460,6 +471,13 @@ Request::Request(asio::io_context& ctx, const std::string& url, std::shared_ptr<
init_default_headers();
}
Request::Request(asio::io_context& ctx, const std::string& url, OnDoneCb onDone, std::shared_ptr<dht::Logger> logger)
: logger_(logger), id_(Request::ids_++), ctx_(ctx), resolver_(std::make_shared<Resolver>(ctx, url, logger))
{
init_default_headers();
add_on_done_callback(std::move(onDone));
}
Request::Request(asio::io_context& ctx, const std::string& host, const std::string& service,
const bool ssl, std::shared_ptr<dht::Logger> logger)
: logger_(logger), id_(Request::ids_++), ctx_(ctx),
@@ -499,7 +517,8 @@ void
Request::init_default_headers()
{
const auto& url = resolver_->get_url();
set_header_field(restinio::http_field_t::host, url.host + ":" + url.service);
set_header_field(restinio::http_field_t::user_agent, "Mozilla/5.0");
set_header_field(restinio::http_field_t::accept, "text/html");
set_target(url.target);
}
@@ -510,39 +529,28 @@ Request::cancel()
c->close();
}
unsigned int
Request::id() const
{
return id_;
}
void
Request::set_connection(std::shared_ptr<Connection> connection)
{
conn_ = connection;
Request::set_connection(std::shared_ptr<Connection> connection) {
conn_ = std::move(connection);
}
std::shared_ptr<Connection>
Request::get_connection() const
{
Request::get_connection() const {
return conn_;
}
void
Request::set_certificate_authority(std::shared_ptr<dht::crypto::Certificate> certificate)
{
Request::set_certificate_authority(std::shared_ptr<dht::crypto::Certificate> certificate) {
server_ca_ = certificate;
}
void
Request::set_identity(const dht::crypto::Identity& identity)
{
Request::set_identity(const dht::crypto::Identity& identity) {
client_identity_ = identity;
}
void
Request::set_logger(std::shared_ptr<dht::Logger> logger)
{
Request::set_logger(std::shared_ptr<dht::Logger> logger) {
logger_ = logger;
}
@@ -553,32 +561,27 @@ Request::set_header(restinio::http_request_header_t header)
}
void
Request::set_method(restinio::http_method_id_t method)
{
Request::set_method(restinio::http_method_id_t method) {
header_.method(method);
}
void
Request::set_target(std::string target)
{
Request::set_target(std::string target) {
header_.request_target(std::move(target));
}
void
Request::set_header_field(restinio::http_field_t field, std::string value)
{
Request::set_header_field(restinio::http_field_t field, std::string value) {
headers_[field] = std::move(value);
}
void
Request::set_connection_type(restinio::http_connection_header_t connection)
{
Request::set_connection_type(restinio::http_connection_header_t connection) {
connection_type_ = connection;
}
void
Request::set_body(std::string body)
{
Request::set_body(std::string body) {
body_ = std::move(body);
}
@@ -597,7 +600,7 @@ void
Request::build()
{
std::stringstream request;
bool append_body = true;
bool append_body = !body_.empty();
// first header
request << header_.method().c_str() << " " << header_.request_target() << " " <<
@@ -613,53 +616,54 @@ Request::build()
// last connection header
const char* conn_str = nullptr;
switch (connection_type_){
case restinio::http_connection_header_t::keep_alive:
conn_str = "keep-alive";
break;
case restinio::http_connection_header_t::upgrade:
if (logger_)
logger_->e("Unsupported connection type 'upgrade', fallback to 'close'");
// fallthrough
case restinio::http_connection_header_t::close:
conn_str = "close";
break;
case restinio::http_connection_header_t::keep_alive:
conn_str = "keep-alive";
conn_str = "close"; // default
break;
}
if (conn_str)
request << "Connection: " << conn_str << "\r\n";
// body & content-length
if (!body_.empty())
request << "Content-Length: " << body_.size() << "\r\n\r\n";
// last delim
if (append_body)
request << body_ << "\r\n";
if (append_body) {
request << "Content-Length: " << body_.size() << "\r\n\r\n"
<< body_;
} else
request << "\r\n";
request_ = request.str();
}
void
Request::add_on_status_callback(OnStatusCb cb)
{
std::lock_guard<std::mutex> lock(cbs_mutex_);
Request::add_on_status_callback(OnStatusCb cb) {
cbs_.on_status = std::move(cb);
}
void
Request::add_on_body_callback(OnDataCb cb)
{
std::lock_guard<std::mutex> lock(cbs_mutex_);
Request::add_on_body_callback(OnDataCb cb) {
cbs_.on_body = std::move(cb);
}
void
Request::add_on_state_change_callback(OnStateChangeCb cb)
{
std::lock_guard<std::mutex> lock(cbs_mutex_);
Request::add_on_state_change_callback(OnStateChangeCb cb) {
cbs_.on_state_change = std::move(cb);
}
void
Request::notify_state_change(const State state)
{
Request::add_on_done_callback(OnDoneCb cb) {
add_on_state_change_callback([onDone=std::move(cb)](State state, const Response& response){
if (state == Request::State::DONE)
onDone(response);
});
}
void
Request::notify_state_change(State state) {
state_ = state;
if (cbs_.on_state_change)
cbs_.on_state_change(state, response_);
@@ -676,38 +680,20 @@ Request::init_parser()
if (!parser_s_)
parser_s_ = std::make_unique<http_parser_settings>();
http_parser_settings_init(parser_s_.get());
{
// user registered callbacks wrappers to store its data in the response
std::lock_guard<std::mutex> lock(cbs_mutex_);
cbs_.on_status = [this, statusCb = std::move(cbs_.on_status)](unsigned int status_code){
response_.status_code = status_code;
if (statusCb)
statusCb(status_code);
};
auto header_field = std::make_shared<std::string>();
cbs_.on_header_field = [header_field, headerFieldCb = std::move(cbs_.on_header_field)](const char* at, size_t length) {
*header_field = std::string(at, length);
if (headerFieldCb)
headerFieldCb(at, length);
};
cbs_.on_header_value = [this, header_field, headerValueCb = std::move(cbs_.on_header_value)](const char* at, size_t length) {
response_.headers[*header_field] = std::string(at, length);
if (headerValueCb)
headerValueCb(at, length);
};
cbs_.on_headers_complete = [this](){
notify_state_change(State::HEADER_RECEIVED);
};
cbs_.on_body = [bodyCb = std::move(cbs_.on_body)](const char* at, size_t length) {
if (bodyCb)
bodyCb(at, length);
};
cbs_.on_message_complete = [this](){
if (logger_)
logger_->d("[http:request:%i] response: message complete", id_);
message_complete_.store(true);
};
}
cbs_.on_status = [this, statusCb = std::move(cbs_.on_status)](unsigned int status_code){
response_.status_code = status_code;
if (statusCb)
statusCb(status_code);
};
auto header_field = std::make_shared<std::string>();
cbs_.on_header_field = [header_field](const char* at, size_t length) {
*header_field = std::string(at, length);
};
cbs_.on_header_value = [this, header_field](const char* at, size_t length) {
response_.headers[*header_field] = std::string(at, length);
};
// http_parser raw c callback (note: no context can be passed into them)
parser_s_->on_status = [](http_parser* parser, const char* /*at*/, size_t /*length*/) -> int {
static_cast<Request*>(parser->data)->cbs_.on_status(parser->status_code);
@@ -722,15 +708,15 @@ Request::init_parser()
return 0;
};
parser_s_->on_body = [](http_parser* parser, const char* at, size_t length) -> int {
static_cast<Request*>(parser->data)->cbs_.on_body(at, length);
static_cast<Request*>(parser->data)->onBody(at, length);
return 0;
};
parser_s_->on_headers_complete = [](http_parser* parser) -> int {
static_cast<Request*>(parser->data)->cbs_.on_headers_complete();
static_cast<Request*>(parser->data)->onHeadersComplete();
return 0;
};
parser_s_->on_message_complete = [](http_parser* parser) -> int {
static_cast<Request*>(parser->data)->cbs_.on_message_complete();
static_cast<Request*>(parser->data)->onComplete();
return 0;
};
}
@@ -748,7 +734,7 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
if (logger_){
std::string eps = "";
for (const auto& endpoint : endpoints)
eps.append(endpoint.address().to_string() + " ");
eps.append(endpoint.address().to_string() + ":" + std::to_string(endpoint.port()) + " ");
logger_->d("[http:request:%i] connect begin: %s", id_, eps.c_str());
}
if (get_url().protocol == "https"){
@@ -772,13 +758,22 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
this_.terminate(ec);
return;
}
else if (ec and this_.logger_)
this_.logger_->e("[http:request:%i] connect: failed with all endpoints", this_.id_);
else {
else if (ec) {
if (this_.logger_)
this_.logger_->d("[http:request:%i] connect success", this_.id_);
this_.logger_->e("[http:request:%i] connect failed with all endpoints: %s", this_.id_, ec.message().c_str());
} else {
// if (this_.logger_)
// this_.logger_->d("[http:request:%i] connect success", this_.id_);
if (this_.get_url().protocol == "https"){
const auto& url = this_.get_url();
auto port = endpoint.port();
if ((url.protocol == "http" && port == (in_port_t)80)
|| (url.protocol == "https" && port == (in_port_t)443))
this_.set_header_field(restinio::http_field_t::host, url.host);
else
this_.set_header_field(restinio::http_field_t::host, url.host + ":" + std::to_string(port));
if (url.protocol == "https") {
if (this_.server_ca_)
this_.conn_->set_ssl_verification(endpoint, asio::ssl::verify_peer
| asio::ssl::verify_fail_if_no_peer_cert);
@@ -789,8 +784,8 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
return;
if (ec and logger)
logger->e("[http:request:%i] handshake error: %s", id, ec.message().c_str());
else if (logger)
logger->d("[http:request:%i] handshake success", id);
//else if (logger)
// logger->d("[http:request:%i] handshake success", id);
if (cb)
cb(ec);
});
@@ -820,7 +815,7 @@ Request::send()
this_.logger_->e("[http:request:%i] resolve error: %s", this_.id_, ec.message().c_str());
this_.terminate(asio::error::connection_aborted);
}
else if (!this_.conn_ or !this_.conn_->is_open()){
else if (!this_.conn_ or !this_.conn_->is_open()) {
this_.connect(std::move(endpoints), [wthis](const asio::error_code &ec) {
if (auto sthis = wthis.lock()) {
if (ec)
@@ -846,10 +841,9 @@ Request::post()
build();
init_parser();
if (logger_){
std::string header; std::getline(std::istringstream(request_), header);
logger_->d("[http:request:%i] send: %s", id_, header.c_str());
}
if (logger_)
logger_->d("[http:request:%i] sending %zu bytes", id_, request_.size());
// write the request to buffer
std::ostream request_stream(&conn_->input());
request_stream << request_;
@@ -870,20 +864,16 @@ Request::terminate(const asio::error_code& ec)
if (finishing_.exchange(true))
return;
if (ec != asio::error::eof and ec != asio::error::operation_aborted and logger_)
logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
// set response outcome, ignore end of file and abort
if ((!ec or ec == asio::error::eof) and parser_)
response_.status_code = parser_->status_code;
else
response_.status_code = 0;
response_.aborted = ec == asio::error::operation_aborted;
if (logger_)
logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
if (connection_type_ != restinio::http_connection_header_t::keep_alive)
if (logger_) {
if (ec and ec != asio::error::eof and ec != asio::error::operation_aborted)
logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
else
logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
}
if (!parser_ or !http_should_keep_alive(parser_.get()))
if (auto c = conn_)
c->close();
notify_state_change(State::DONE);
@@ -900,41 +890,63 @@ Request::handle_request(const asio::error_code& ec)
terminate(asio::error::not_connected);
return;
}
if (logger_)
logger_->d("[http:request:%i] send success", id_);
// if (logger_)
// logger_->d("[http:request:%i] send success", id_);
// read response
notify_state_change(State::RECEIVING);
std::weak_ptr<Request> wthis = shared_from_this();
conn_->async_read_until(HTTP_HEADER_DELIM, [wthis](const asio::error_code& ec, size_t){
conn_->async_read_until(HTTP_HEADER_DELIM, [wthis](const asio::error_code& ec, size_t n_bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_header(ec);
sthis->handle_response(ec, n_bytes);
});
}
void
Request::handle_response_header(const asio::error_code& ec)
Request::handle_response(const asio::error_code& ec, size_t n_bytes)
{
if (ec && ec != asio::error::eof){
terminate(ec);
return;
}
if (!conn_->is_open()){
terminate(asio::error::not_connected);
auto request = (ec == asio::error::eof) ? std::string{} : conn_->read_bytes();
size_t ret = http_parser_execute(parser_.get(), parser_s_.get(), request.c_str(), request.size());
if (ret != request.size()) {
if (logger_)
logger_->e("Error parsing HTTP: %zu %s %s", ret,
http_errno_name(HTTP_PARSER_ERRNO(parser_)),
http_errno_description(HTTP_PARSER_ERRNO(parser_)));
terminate(asio::error::basic_errors::broken_pipe);
return;
}
if (logger_)
logger_->d("[http:request:%i] response headers received", id_);
// read the header
std::string header;
std::string headers;
while (std::getline(conn_->data(), header) && header != "\r"){
headers.append(header + "\n");
}
headers.append("\n");
parse_request(headers);
std::weak_ptr<Request> wthis = shared_from_this();
if (state_ != State::DONE and parser_ and not http_body_is_final(parser_.get())) {
auto toRead = parser_->content_length ? std::min<uint64_t>(parser_->content_length, 64 * 1024) : 64 * 1024;
std::weak_ptr<Request> wthis = shared_from_this();
conn_->async_read_some(toRead, [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response(ec, bytes);
});
}
}
void
Request::onBody(const char* at, size_t length)
{
if (cbs_.on_body)
cbs_.on_body(at, length);
else
response_.body.insert(response_.body.end(), at, at+length);
}
void
Request::onComplete() {
terminate(asio::error::eof);
}
void
Request::onHeadersComplete() {
notify_state_change(State::HEADER_RECEIVED);
auto expect_it = headers_.find(restinio::http_field_t::expect);
if (expect_it != headers_.end() and (expect_it->second == "100-continue") and response_.status_code != 200){
@@ -942,167 +954,13 @@ Request::handle_response_header(const asio::error_code& ec)
request_.append(body_);
std::ostream request_stream(&conn_->input());
request_stream << body_ << "\r\n";
std::weak_ptr<Request> wthis = shared_from_this();
conn_->async_write([wthis](const asio::error_code& ec, size_t) {
if (auto sthis = wthis.lock())
sthis->handle_request(ec);
});
return;
}
auto connection_it = response_.headers.find(HTTP_HEADER_CONNECTION);
auto content_length_it = response_.headers.find(HTTP_HEADER_CONTENT_LENGTH);
auto transfer_encoding_it = response_.headers.find(HTTP_HEADER_TRANSFER_ENCODING);
// has content-length
if (content_length_it != response_.headers.end())
{
unsigned int content_length = atoi(content_length_it->second.c_str());
response_.body.append(conn_->read_bytes(content_length));
// full body already in the header
if (response_.body.size() + 1 == content_length) {
response_.body.append("\n");
parse_request(response_.body);
if (message_complete_.load())
terminate(asio::error::eof);
}
else { // more chunks to come (don't add the missing \n from std::getline)
conn_->async_read(content_length - response_.body.size(), [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_body(ec, bytes);
});
}
}
// server wants to keep sending or we have content-length defined
else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_KEEP_ALIVE)
{
conn_->async_read_until(BODY_VALUE_DELIM, [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_body(ec, bytes);
});
}
// server wants to close the connection
else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_CLOSE)
{
terminate(asio::error::eof);
}
// client wants to close the connection
else if (connection_type_ == restinio::http_connection_header_t::close)
{
terminate(asio::error::eof);
}
else if (transfer_encoding_it != response_.headers.end() and
transfer_encoding_it->second == HTTP_HEADER_TRANSFER_ENCODING_CHUNKED)
{
std::string chunk_size;
std::getline(conn_->data(), chunk_size);
unsigned int content_length = std::stoul(chunk_size, nullptr, 16);
conn_->async_read(content_length, [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_body(ec, bytes);
});
}
}
void
Request::handle_response_body(const asio::error_code& ec, size_t bytes)
{
if (ec && ec != asio::error::eof){
terminate(ec);
return;
}
if (!conn_->is_open()){
terminate(asio::error::not_connected);
return;
}
if (logger_)
logger_->d("[http:request:%i] response body: %i bytes received", id_, bytes);
if (bytes == 0){
terminate(asio::error::eof);
return;
}
// avoid creating non-existant headers by accessing the headers map without the presence of key
auto connection_it = response_.headers.find(HTTP_HEADER_CONNECTION);
auto content_length_it = response_.headers.find(HTTP_HEADER_CONTENT_LENGTH);
auto transfer_encoding_it = response_.headers.find(HTTP_HEADER_TRANSFER_ENCODING);
// read the content-length body
unsigned int content_length;
if (content_length_it != response_.headers.end() and !response_.body.empty()){
// extract the content-length
content_length = atoi(content_length_it->second.c_str());
response_.body.append(conn_->read_bytes(bytes));
// check if fully parsed
if (response_.body.size() == content_length)
parse_request(response_.body);
}
// read and parse the chunked encoding fragment
else {
auto body = conn_->read_until(BODY_VALUE_DELIM);
response_.body += body;
if (body == "0\r\n"){
parse_request(response_.body);
terminate(asio::error::eof);
return;
}
parse_request(body + '\n');
}
std::weak_ptr<Request> wthis = shared_from_this();
// should be executed after each parse_request who can trigger http_parser on_message_complete
if (message_complete_.load()){
terminate(asio::error::eof);
}
// has content-length
else if (content_length_it != response_.headers.end() and response_.body.size() != content_length)
{
conn_->async_read(content_length - response_.body.size(), [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_body(ec, bytes);
});
}
// server wants to keep sending
else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_KEEP_ALIVE)
{
conn_->async_read_until(BODY_VALUE_DELIM, [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_body(ec, bytes);
});
}
// server wants to close the connection
else if (connection_it != response_.headers.end() and connection_it->second == HTTP_HEADER_CONNECTION_CLOSE)
{
terminate(asio::error::eof);
}
// client wants to close the connection
else if (connection_type_ == restinio::http_connection_header_t::close)
{
terminate(asio::error::eof);
}
else if (transfer_encoding_it != response_.headers.end() and
transfer_encoding_it->second == HTTP_HEADER_TRANSFER_ENCODING_CHUNKED)
{
std::string chunk_size;
std::getline(conn_->data(), chunk_size);
if (chunk_size.size() == 0){
parse_request(response_.body);
terminate(asio::error::eof);
}
else
conn_->async_read_until(BODY_VALUE_DELIM, [wthis](const asio::error_code& ec, size_t bytes){
if (auto sthis = wthis.lock())
sthis->handle_response_body(ec, bytes);
});
}
}
size_t
Request::parse_request(const std::string& request)
{
std::lock_guard<std::mutex> lock(cbs_mutex_);
return http_parser_execute(parser_.get(), parser_s_.get(), request.c_str(), request.size());
}
} // namespace http
+10 -10
View File
@@ -56,9 +56,9 @@ printLog(std::ostream& s, char const *m, va_list args) {
s << std::endl;
}
std::unique_ptr<Logger>
std::shared_ptr<Logger>
getStdLogger() {
return std::unique_ptr<Logger>(new Logger(
return std::make_shared<Logger>(
[](char const *m, va_list args) {
std::cerr << red;
printLog(std::cerr, m, args);
@@ -70,22 +70,22 @@ getStdLogger() {
std::cout << def;
},
[](char const *m, va_list args) { printLog(std::cout, m, args); }
));
);
}
std::unique_ptr<Logger>
std::shared_ptr<Logger>
getFileLogger(const std::string &path) {
auto logfile = std::make_shared<std::ofstream>();
logfile->open(path, std::ios::out);
return std::unique_ptr<Logger>(new Logger(
return std::make_shared<Logger>(
[=](char const *m, va_list args) { printLog(*logfile, m, args); },
[=](char const *m, va_list args) { printLog(*logfile, m, args); },
[=](char const *m, va_list args) { printLog(*logfile, m, args); }
));
);
}
std::unique_ptr<Logger>
std::shared_ptr<Logger>
getSyslogLogger(const char* name) {
#ifndef _WIN32
struct Syslog {
@@ -103,13 +103,13 @@ getSyslogLogger(const char* name) {
logfile = std::make_shared<Syslog>(name);
opened_logfile = logfile;
}
return std::unique_ptr<Logger>(new Logger(
return std::make_shared<Logger>(
[logfile](char const *m, va_list args) { vsyslog(LOG_ERR, m, args); },
[logfile](char const *m, va_list args) { vsyslog(LOG_WARNING, m, args); },
[logfile](char const *m, va_list args) { vsyslog(LOG_INFO, m, args); }
));
);
#else
return std::unique_ptr<Logger>(new Logger());
return std::make_shared<Logger>();
#endif
}
+126 -352
View File
@@ -2,6 +2,7 @@
* Copyright (C) 2014-2019 Savoir-faire Linux Inc.
* Author(s) : Mingrui Zhang <mingrui.zhang@savoirfairelinux.com>
* Vsevolod Ivanov <vsevolod.ivanov@savoirfairelinux.com>
* Adrien Béraud <adrien.beraud@savoirfairelinux.com>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
@@ -21,402 +22,166 @@
#include "network_utils.h"
#include "utils.h"
#ifdef _WIN32
#include <Ws2tcpip.h> // needed for ip_mreq definition for multicast
#include <Windows.h>
#include <cstring>
#if defined(_MSC_VER)
#include <BaseTsd.h>
typedef SSIZE_T ssize_t;
#endif
#define close(x) closesocket(x)
#else
#include <sys/types.h>
#include <unistd.h>
#endif
#include <fcntl.h>
#include <asio.hpp>
#ifndef IPV6_JOIN_GROUP
#define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP
#endif
using namespace std::chrono_literals;
namespace dht {
// Organization-local Scope multicast
constexpr char MULTICAST_ADDRESS_IPV4[12] = "239.192.0.1";
constexpr char MULTICAST_ADDRESS_IPV6[10] = "ff08::101";
constexpr char MULTICAST_ADDRESS_IPV4[] = "239.192.0.1";
constexpr char MULTICAST_ADDRESS_IPV6[] = "ff08::101";
class PeerDiscovery::DomainPeerDiscovery
{
public:
DomainPeerDiscovery(sa_family_t domain, in_port_t port);
DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext = {}, Sp<Logger> logger = {});
~DomainPeerDiscovery();
/**
* startDiscovery - Keep Listening data from the sender until node is joinned or stop is called
*/
void startDiscovery(const std::string &type, ServiceDiscoveredCallback callback);
/**
* startPublish - Keeping sending data until node is joinned or stop is called - msgpack
*/
void startPublish(const std::string &type, const msgpack::sbuffer &pack_buf);
/**
* 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);
/**
* Configure the sockopt to be able to listen multicast group
*/
static void joinMulticast(int sockfd, sa_family_t family);
/**
* Join the threads
*/
void join() {
if(listening_.joinable()) listening_.join();
if(sending_.joinable()) sending_.join();
}
private:
Sp<Logger> logger_;
//dmtx_ for callbackmap_ and drunning_ (write)
std::mutex dmtx_;
//mtx_ for messages_ and lrunning (listen)
std::mutex mtx_;
std::condition_variable cv_;
bool lrunning_ {false};
bool drunning_ {false};
sa_family_t domain_ {AF_UNSPEC};
int port_;
int sockFd_ {-1};
int stopWriteFd_ {-1};
std::shared_ptr<asio::io_context> ioContext_;
asio::ip::udp::socket sockFd_;
asio::ip::udp::endpoint sockAddrSend_;
asio::steady_timer publishTimer_;
SockAddr sockAddrSend_;
//Thread export to be joined
std::thread listening_;
std::thread sending_;
std::array<char, 64 * 1024> receiveBuf_;
asio::ip::udp::endpoint receiveFrom_;
msgpack::sbuffer sbuf_;
std::map<std::string, msgpack::sbuffer> messages_;
std::map<std::string, ServiceDiscoveredCallback> callbackmap_;
bool lrunning_ {false};
bool drunning_ {false};
/**
* Multicast Socket Initialization, accept IPV4, IPV6
*/
static int init(sa_family_t domain);
/**
* Receive messages
*/
std::pair<SockAddr, msgpack::object_handle> recvFrom();
/**
* Listener pack thread loop
*/
void loopListener();
/**
* Listener Parameters Setup
*/
void setupListener();
/**
* Sender Parameters Setup
*/
void setupSender();
/**
* Sender Parameters Setup
*/
void reloadMessages();
void stopDiscovery();
void stopPublish();
void publish();
};
PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(sa_family_t domain, in_port_t port)
: domain_(domain), port_(port), sockFd_(init(domain))
PeerDiscovery::DomainPeerDiscovery::DomainPeerDiscovery(asio::ip::udp domain, in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
: logger_(logger)
, ioContext_(ioContext)
, sockFd_(*ioContext_, domain)
, sockAddrSend_(asio::ip::address::from_string(domain.family() == AF_INET ? MULTICAST_ADDRESS_IPV4
: MULTICAST_ADDRESS_IPV6), port)
, publishTimer_(*ioContext)
{
joinMulticast(sockFd_, domain);
setupListener();
setupSender();
sockFd_.set_option(asio::ip::multicast::join_group(sockAddrSend_.address()));
sockFd_.set_option(asio::ip::udp::socket::reuse_address(true));
sockFd_.bind({domain, port});
}
PeerDiscovery::DomainPeerDiscovery::~DomainPeerDiscovery()
{
stop();
join();
if (sockFd_ != -1)
close(sockFd_);
#ifdef _WIN32
WSACleanup();
#endif
}
int
PeerDiscovery::DomainPeerDiscovery::init(sa_family_t domain)
{
#ifdef _WIN32
WSADATA wsaData;
if (WSAStartup(0x0101, &wsaData)) {
throw std::runtime_error(std::string("Can't initialize Winsock2 ") + strerror(errno));
}
#endif
int sockfd = socket(domain, SOCK_DGRAM, 0);
if (sockfd < 0) {
throw std::runtime_error(std::string("Socket Creation Error: ") + strerror(errno));
}
net::setNonblocking(sockfd);
return sockfd;
}
void
PeerDiscovery::DomainPeerDiscovery::setupListener()
{
SockAddr sockAddrListen_;
sockAddrListen_.setFamily(domain_);
sockAddrListen_.setPort(port_);
sockAddrListen_.setAny();
unsigned int opt = 1;
if (setsockopt(sockFd_, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, sizeof(opt)) < 0) {
std::cerr << "setsockopt SO_REUSEADDR failed: " << strerror(errno) << std::endl;
}
#ifdef SO_REUSEPORT
if (setsockopt(sockFd_, SOL_SOCKET, SO_REUSEPORT, (char*)&opt, sizeof(opt)) < 0) {
std::cerr << "setsockopt SO_REUSEPORT failed: " << strerror(errno) << std::endl;
}
#endif
// bind to receive address
if (bind(sockFd_, sockAddrListen_.get(), sockAddrListen_.getLength()) < 0){
throw std::runtime_error(std::string("Error binding socket: ") + strerror(errno));
}
}
void
PeerDiscovery::DomainPeerDiscovery::joinMulticast(int sockfd, sa_family_t family)
{
switch (family)
{
case AF_INET:{
ip_mreq config_ipv4;
//This option can be used to set the interface for sending outbound
//multicast datagrams from the sockets application.
config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_IF, (char*)&config_ipv4.imr_interface, sizeof( struct in_addr )) < 0 ) {
throw std::runtime_error(std::string("Bound Network Interface IPv4 Error: ") + strerror(errno));
}
//The IP_MULTICAST_TTL socket option allows the application to primarily
//limit the lifetime of the packet in the Internet and prevent it from circulating indefinitely
unsigned char ttl4 = 20;
if( setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, (char*)&ttl4, sizeof( ttl4 )) < 0 ) {
throw std::runtime_error(std::string("TTL Sockopt Error: ") + strerror(errno));
}
// config the listener to be interested in joining in the multicast group
config_ipv4.imr_multiaddr.s_addr = inet_addr(MULTICAST_ADDRESS_IPV4);
config_ipv4.imr_interface.s_addr = htonl(INADDR_ANY);
if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*)&config_ipv4, sizeof(config_ipv4)) < 0){
throw std::runtime_error(std::string(" Member Addition IPv4 Error: ") + strerror(errno));
}
break;
}
case AF_INET6: {
ipv6_mreq config_ipv6;
/* unsigned int outif = 0;
if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &outif, sizeof(outif)) < 0) {
std::cerr << "Can't assign multicast interface: " << strerror(errno) << std::endl;
} */
unsigned int ttl6 = 20;
if( setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, (char*)&ttl6, sizeof( ttl6 )) < 0 ) {
throw std::runtime_error(std::string("Hop Count Set Error: ") + strerror(errno));
}
config_ipv6.ipv6mr_interface = 0;
inet_pton(AF_INET6, MULTICAST_ADDRESS_IPV6, &config_ipv6.ipv6mr_multiaddr);
if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_JOIN_GROUP, (char*)&config_ipv6, sizeof(config_ipv6)) < 0){
throw std::runtime_error(std::string("Member Addition IPv6 Error: ") + strerror(errno));
}
break;
}
}
sockFd_.close();
}
void
PeerDiscovery::DomainPeerDiscovery::startDiscovery(const std::string &type, ServiceDiscoveredCallback callback)
{
std::unique_lock<std::mutex> lck(dmtx_);
std::lock_guard<std::mutex> lck(dmtx_);
callbackmap_[type] = callback;
if (not drunning_) {
if (listening_.joinable())
listening_.join();
drunning_ = true;
listening_ = std::thread(&DomainPeerDiscovery::loopListener, this);
ioContext_->post(std::bind(&PeerDiscovery::DomainPeerDiscovery::loopListener, this));
}
}
std::pair<SockAddr, msgpack::object_handle>
PeerDiscovery::DomainPeerDiscovery::recvFrom()
{
sockaddr_storage sa;
socklen_t sa_len = sizeof(sa);
std::array<uint8_t, 64 * 1024> buf;
ssize_t nbytes = recvfrom(
sockFd_,
(char*)buf.data(), buf.size(),
0,
(sockaddr*)&sa, &sa_len
);
if (nbytes < 0)
throw std::runtime_error(std::string("Error receiving packet: ") + strerror(errno));
return { SockAddr { sa, sa_len }, msgpack::unpack(reinterpret_cast<char*>(buf.data()), nbytes) };
}
void
PeerDiscovery::DomainPeerDiscovery::loopListener()
{
int stopfds_pipe[2];
#ifndef _WIN32
if (pipe(stopfds_pipe) == -1)
throw std::runtime_error(std::string("Can't open pipe: ") + strerror(errno));
#else
net::udpPipe(stopfds_pipe);
#endif
int stop_readfd = stopfds_pipe[0];
stopWriteFd_ = stopfds_pipe[1];
while (true) {
{
std::lock_guard<std::mutex> lck(dmtx_);
if (not drunning_)
break;
std::lock_guard<std::mutex> lck(dmtx_);
if (not drunning_)
return;
sockFd_.async_receive_from(asio::buffer(receiveBuf_), receiveFrom_, [this](const asio::error_code& error, size_t bytes) {
if (error == asio::error::operation_aborted)
return;
if (error) {
if (logger_)
logger_->e("Error receiving message: %s", error.message().c_str());
}
try {
auto rcv = msgpack::unpack(receiveBuf_.data(), bytes);
msgpack::object obj = rcv.get();
fd_set readfds;
FD_ZERO(&readfds);
FD_SET(stop_readfd, &readfds);
FD_SET(sockFd_, &readfds);
int data_coming = select(std::max(sockFd_, stop_readfd) + 1, &readfds, nullptr, nullptr, nullptr);
{
std::lock_guard<std::mutex> lck(dmtx_);
if (not drunning_)
break;
}
if (data_coming < 0) {
if(errno != EINTR) {
perror("Select Error");
std::this_thread::sleep_for( std::chrono::seconds(1) );
}
} else if (data_coming > 0) {
if (FD_ISSET(stop_readfd, &readfds)) {
std::array<uint8_t, 64 * 1024> buf;
recv(stop_readfd, (char*)buf.data(), buf.size(), 0);
}
try {
auto rcv = recvFrom();
msgpack::object obj = rcv.second.get();
if (obj.type != msgpack::type::MAP)
if (obj.type != msgpack::type::MAP)
throw msgpack::type_error{};
for (unsigned i = 0; i < obj.via.map.size; i++) {
auto& o = obj.via.map.ptr[i];
if (o.key.type != msgpack::type::STR)
continue;
for (unsigned i = 0; i < obj.via.map.size; i++) {
auto& o = obj.via.map.ptr[i];
if (o.key.type != msgpack::type::STR)
continue;
auto key = o.key.as<std::string>();
ServiceDiscoveredCallback cb;
{
std::lock_guard<std::mutex> lck(dmtx_);
auto key = o.key.as<std::string>();
ServiceDiscoveredCallback cb;
{
std::lock_guard<std::mutex> lck(dmtx_);
if (drunning_) {
auto callback = callbackmap_.find(key);
if (callback != callbackmap_.end())
cb = callback->second;
}
if (cb)
cb(std::move(o.val), std::move(rcv.first));
} else
return;
}
} catch (const std::exception& e) {
std::cerr << "Error receiving packet: " << e.what() << std::endl;
if (cb)
cb(std::move(o.val), SockAddr{ receiveFrom_.data(), (socklen_t)receiveFrom_.size() });
}
} catch (const std::exception& e) {
if (logger_)
logger_->e("Error receiving packet: %s", e.what());
}
}
if (stop_readfd != -1)
close(stop_readfd);
if (stopWriteFd_ != -1) {
close(stopWriteFd_);
stopWriteFd_ = -1;
}
loopListener();
});
}
void
PeerDiscovery::DomainPeerDiscovery::setupSender()
PeerDiscovery::DomainPeerDiscovery::publish()
{
// Setup sender address
sockAddrSend_.setFamily(domain_);
sockAddrSend_.setAddress(domain_ == AF_INET ? MULTICAST_ADDRESS_IPV4 : MULTICAST_ADDRESS_IPV6);
sockAddrSend_.setPort(port_);
std::lock_guard<std::mutex> lck(mtx_);
if (lrunning_) {
sockFd_.async_send_to(asio::buffer(sbuf_.data(), sbuf_.size()), sockAddrSend_, [logger=logger_](const asio::error_code& ec, size_t){
if (ec and ec != asio::error::operation_aborted) {
if (logger)
logger->w("Error sending packet: %s", ec.message().c_str());
}
});
publishTimer_.expires_after(3s);
publishTimer_.async_wait(std::bind(&PeerDiscovery::DomainPeerDiscovery::publish, this));
}
}
void
PeerDiscovery::DomainPeerDiscovery::startPublish(const std::string &type, const msgpack::sbuffer &pack_buf)
{
//Set up New Sending pack
msgpack::sbuffer pack_buf_c;
pack_buf_c.write(pack_buf.data(),pack_buf.size());
std::unique_lock<std::mutex> lck(mtx_);
std::lock_guard<std::mutex> lck(mtx_);
messages_[type] = std::move(pack_buf_c);
reloadMessages();
if (not lrunning_) {
if (sending_.joinable())
sending_.join();
lrunning_ = true;
sending_ = std::thread([this](){
std::unique_lock<std::mutex> lck(mtx_);
while (lrunning_) {
ssize_t nbytes = sendto(
sockFd_,
sbuf_.data(),
sbuf_.size(),
0,
sockAddrSend_.get(),
sockAddrSend_.getLength()
);
if (nbytes < 0) {
std::cerr << "Error sending packet: " << strerror(errno) << std::endl;
}
if (cv_.wait_for(lck,std::chrono::seconds(3),[&]{ return !lrunning_; }))
break;
}
});
ioContext_->post(std::bind(&PeerDiscovery::DomainPeerDiscovery::publish, this));
}
}
bool
PeerDiscovery::DomainPeerDiscovery::stopDiscovery(const std::string &type)
PeerDiscovery::DomainPeerDiscovery::stopDiscovery(const std::string& type)
{
std::lock_guard<std::mutex> lck(dmtx_);
if (callbackmap_.erase(type) > 0) {
@@ -428,7 +193,7 @@ PeerDiscovery::DomainPeerDiscovery::stopDiscovery(const std::string &type)
}
bool
PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string &type)
PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string& type)
{
std::lock_guard<std::mutex> lck(mtx_);
if (messages_.erase(type) > 0) {
@@ -444,20 +209,8 @@ PeerDiscovery::DomainPeerDiscovery::stopPublish(const std::string &type)
void
PeerDiscovery::DomainPeerDiscovery::stopDiscovery()
{
#ifdef _WIN32
#define write(s, b, f) send((s), (b), (f), 0)
#endif
if (drunning_) {
if (drunning_)
drunning_ = false;
if (stopWriteFd_ != -1) {
if (write(stopWriteFd_, "\0", 1) == -1) {
std::cerr << "Can't send stop message: " << strerror(errno) << std::endl;
}
}
}
#ifdef _WIN32
#undef write
#endif
}
void
@@ -465,7 +218,7 @@ PeerDiscovery::DomainPeerDiscovery::stopPublish()
{
if (lrunning_) {
lrunning_ = false;
cv_.notify_all();
publishTimer_.cancel();
}
}
@@ -473,11 +226,11 @@ void
PeerDiscovery::DomainPeerDiscovery::stop()
{
{
std::unique_lock<std::mutex> lck(dmtx_);
std::lock_guard<std::mutex> lck(dmtx_);
stopDiscovery();
}
{
std::unique_lock<std::mutex> lck(mtx_);
std::lock_guard<std::mutex> lck(mtx_);
stopPublish();
}
}
@@ -494,38 +247,65 @@ PeerDiscovery::DomainPeerDiscovery::reloadMessages()
}
}
PeerDiscovery::PeerDiscovery(in_port_t port)
PeerDiscovery::PeerDiscovery(in_port_t port, Sp<asio::io_context> ioContext, Sp<Logger> logger)
{
if (not ioContext) {
ioContext = std::make_shared<asio::io_context>();
ioContext_ = ioContext;
ioRunnner_ = std::thread([logger, ioContext] {
try {
if (logger)
logger->d("[peerdiscovery] starting io_context");
auto work = asio::make_work_guard(*ioContext);
ioContext->run();
if (logger)
logger->d("[peerdiscovery] io_context stopped");
}
catch (const std::exception& ex){
if (logger)
logger->e("[peerdiscovery] run error: %s", ex.what());
}
});
}
try {
peerDiscovery4_.reset(new DomainPeerDiscovery(AF_INET, port));
peerDiscovery4_.reset(new DomainPeerDiscovery(asio::ip::udp::v4(), port, ioContext, logger));
} catch(const std::exception& e){
std::cerr << "Can't start peer discovery (IPv4): " << e.what() << std::endl;
if (logger)
logger->e("[peerdiscovery] can't start IPv4: %s", e.what());
}
try {
peerDiscovery6_.reset(new DomainPeerDiscovery(AF_INET6, port));
peerDiscovery6_.reset(new DomainPeerDiscovery(asio::ip::udp::v6(), port, ioContext, logger));
} catch(const std::exception& e) {
std::cerr << "Can't start peer discovery (IPv6): " << e.what() << std::endl;
if (logger)
logger->e("[peerdiscovery] can't start IPv6: %s", e.what());
}
}
PeerDiscovery::~PeerDiscovery(){}
void
PeerDiscovery::startDiscovery(const std::string &type, ServiceDiscoveredCallback callback)
{
if(peerDiscovery4_) peerDiscovery4_->startDiscovery(type, callback);
if(peerDiscovery6_) peerDiscovery6_->startDiscovery(type, callback);
PeerDiscovery::~PeerDiscovery() {
stop();
if (ioContext_)
ioContext_->stop();
if (ioRunnner_.joinable())
ioRunnner_.join();
}
void
PeerDiscovery::startPublish(const std::string &type, const msgpack::sbuffer &pack_buf)
PeerDiscovery::startDiscovery(const std::string& type, ServiceDiscoveredCallback callback)
{
if(peerDiscovery4_) peerDiscovery4_->startPublish(type, pack_buf);
if(peerDiscovery6_) peerDiscovery6_->startPublish(type, pack_buf);
if (peerDiscovery4_) peerDiscovery4_->startDiscovery(type, callback);
if (peerDiscovery6_) peerDiscovery6_->startDiscovery(type, callback);
}
void
PeerDiscovery::startPublish(sa_family_t domain, const std::string &type, const msgpack::sbuffer &pack_buf)
PeerDiscovery::startPublish(const std::string& type, const msgpack::sbuffer& pack_buf)
{
if (peerDiscovery4_) peerDiscovery4_->startPublish(type, pack_buf);
if (peerDiscovery6_) peerDiscovery6_->startPublish(type, pack_buf);
}
void
PeerDiscovery::startPublish(sa_family_t domain, const std::string& type, const msgpack::sbuffer& pack_buf)
{
if (domain == AF_INET) {
if (peerDiscovery4_) peerDiscovery4_->startPublish(type, pack_buf);
@@ -537,8 +317,8 @@ PeerDiscovery::startPublish(sa_family_t domain, const std::string &type, const m
void
PeerDiscovery::stop()
{
if(peerDiscovery4_) peerDiscovery4_->stop();
if(peerDiscovery6_) peerDiscovery6_->stop();
if (peerDiscovery4_) peerDiscovery4_->stop();
if (peerDiscovery6_) peerDiscovery6_->stop();
}
bool
@@ -558,7 +338,7 @@ PeerDiscovery::stopPublish(const std::string &type)
}
bool
PeerDiscovery::stopPublish(sa_family_t domain, const std::string &type)
PeerDiscovery::stopPublish(sa_family_t domain, const std::string& type)
{
if (domain == AF_INET) {
return peerDiscovery4_ and peerDiscovery4_->stopPublish(type);
@@ -568,10 +348,4 @@ PeerDiscovery::stopPublish(sa_family_t domain, const std::string &type)
return false;
}
void
PeerDiscovery::join() {
if(peerDiscovery4_) peerDiscovery4_->join();
if(peerDiscovery6_) peerDiscovery6_->join();
}
}
+3
View File
@@ -17,6 +17,9 @@
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "securedht.h"
#include "rng.h"
+17 -3
View File
@@ -246,7 +246,7 @@ std::time_t to_time_t(time_point t) {
}
Blob
unpackBlob(msgpack::object& o) {
unpackBlob(const msgpack::object& o) {
switch (o.type) {
case msgpack::type::BIN:
return {o.via.bin.ptr, o.via.bin.ptr+o.via.bin.size};
@@ -270,11 +270,25 @@ unpackMsg(Blob b) {
}
msgpack::object*
findMapValue(msgpack::object& map, const std::string& key) {
findMapValue(const msgpack::object& map, const char* key) {
if (map.type != msgpack::type::MAP) throw msgpack::type_error();
for (unsigned i = 0; i < map.via.map.size; i++) {
auto& o = map.via.map.ptr[i];
if (o.key.type == msgpack::type::STR && o.key.as<std::string>() == key)
if (o.key.type == msgpack::type::STR
&& std::strncmp(o.key.via.str.ptr, key, o.key.via.str.size) == 0)
return &o.val;
}
return nullptr;
}
msgpack::object*
findMapValue(const msgpack::object& map, const std::string& key) {
if (map.type != msgpack::type::MAP) throw msgpack::type_error();
for (unsigned i = 0; i < map.via.map.size; i++) {
auto& o = map.via.map.ptr[i];
if (o.key.type == msgpack::type::STR
&& key.size() == o.key.via.str.size
&& std::strncmp(o.key.via.str.ptr, key.data(), o.key.via.str.size) == 0)
return &o.val;
}
return nullptr;
+71 -77
View File
@@ -17,6 +17,10 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "value.h"
#include "default_types.h"
@@ -83,20 +87,6 @@ ValueType::DEFAULT_STORE_POLICY(InfoHash, const std::shared_ptr<Value>& v, const
return v->size() <= MAX_VALUE_SIZE;
}
msgpack::object*
findMapValue(const msgpack::object& map, const std::string& key) {
if (map.type != msgpack::type::MAP) throw msgpack::type_error();
for (unsigned i = 0; i < map.via.map.size; i++) {
auto& o = map.via.map.ptr[i];
if(o.key.type != msgpack::type::STR)
continue;
if (o.key.as<std::string>() == key) {
return &o.val;
}
}
return nullptr;
}
size_t
Value::size() const
{
@@ -104,20 +94,24 @@ Value::size() const
}
void
Value::msgpack_unpack(msgpack::object o)
Value::msgpack_unpack(const msgpack::object& o)
{
if (o.type != msgpack::type::MAP) throw msgpack::type_error();
if (o.via.map.size < 2) throw msgpack::type_error();
if (o.type != msgpack::type::MAP or o.via.map.size < 2)
throw msgpack::type_error();
if (auto rid = findMapValue(o, "id")) {
if (auto rid = findMapValue(o, VALUE_KEY_ID)) {
id = rid->as<Id>();
} else
throw msgpack::type_error();
if (auto rdat = findMapValue(o, "dat")) {
if (auto rdat = findMapValue(o, VALUE_KEY_DAT)) {
msgpack_unpack_body(*rdat);
} else
throw msgpack::type_error();
if (auto rprio = findMapValue(o, VALUE_KEY_PRIO)) {
priority = rprio->as<unsigned>();
}
}
void
@@ -136,37 +130,37 @@ Value::msgpack_unpack_body(const msgpack::object& o)
} else {
if (o.type != msgpack::type::MAP)
throw msgpack::type_error();
auto rbody = findMapValue(o, "body");
auto rbody = findMapValue(o, VALUE_KEY_BODY);
if (not rbody)
throw msgpack::type_error();
if (auto rdata = findMapValue(*rbody, "data")) {
if (auto rdata = findMapValue(*rbody, VALUE_KEY_DATA)) {
data = unpackBlob(*rdata);
} else
throw msgpack::type_error();
if (auto rtype = findMapValue(*rbody, "type")) {
if (auto rtype = findMapValue(*rbody, VALUE_KEY_TYPE)) {
type = rtype->as<ValueType::Id>();
} else
throw msgpack::type_error();
if (auto rutype = findMapValue(*rbody, "utype")) {
if (auto rutype = findMapValue(*rbody, VALUE_KEY_USERTYPE)) {
user_type = rutype->as<std::string>();
}
if (auto rowner = findMapValue(*rbody, "owner")) {
if (auto rseq = findMapValue(*rbody, "seq"))
if (auto rowner = findMapValue(*rbody, VALUE_KEY_OWNER)) {
if (auto rseq = findMapValue(*rbody, VALUE_KEY_SEQ))
seq = rseq->as<decltype(seq)>();
else
throw msgpack::type_error();
crypto::PublicKey new_owner;
new_owner.msgpack_unpack(*rowner);
owner = std::make_shared<const crypto::PublicKey>(std::move(new_owner));
if (auto rrecipient = findMapValue(*rbody, "to")) {
if (auto rrecipient = findMapValue(*rbody, VALUE_KEY_TO)) {
recipient = rrecipient->as<InfoHash>();
}
if (auto rsig = findMapValue(o, "sig")) {
if (auto rsig = findMapValue(o, VALUE_KEY_SIGNATURE)) {
signature = unpackBlob(*rsig);
} else
throw msgpack::type_error();
@@ -175,63 +169,65 @@ Value::msgpack_unpack_body(const msgpack::object& o)
}
#ifdef OPENDHT_JSONCPP
Value::Value(Json::Value& json)
Value::Value(const Json::Value& json)
{
id = Value::Id(unpackId(json, "id"));
if (json.isMember("cypher")) {
auto cypherStr = json["cypher"].asString();
cypherStr = base64_decode(cypherStr);
cypher = std::vector<unsigned char>(cypherStr.begin(), cypherStr.end());
}
if (json.isMember("sig")) {
auto sigStr = json["sig"].asString();
sigStr = base64_decode(sigStr);
signature = std::vector<unsigned char>(sigStr.begin(), sigStr.end());
}
if (json.isMember("seq"))
seq = json["seq"].asInt();
if (json.isMember("owner")) {
auto ownerStr = json["owner"].asString();
id = Value::Id(unpackId(json, VALUE_KEY_ID));
const auto& jcypher = json["cypher"];
if (jcypher.isString())
cypher = base64_decode(jcypher.asString());
const auto& jsig = json[VALUE_KEY_SIGNATURE];
if (jsig.isString())
signature = base64_decode(jsig.asString());
const auto& jseq = json[VALUE_KEY_SEQ];
if (!jseq.isNull())
seq = jseq.asInt();
const auto& jowner = json[VALUE_KEY_OWNER];
if (jowner.isString()) {
auto ownerStr = jowner.asString();
auto ownerBlob = std::vector<unsigned char>(ownerStr.begin(), ownerStr.end());
owner = std::make_shared<const crypto::PublicKey>(ownerBlob);
}
if (json.isMember("to")) {
auto toStr = json["to"].asString();
recipient = InfoHash(toStr);
}
if (json.isMember("type"))
type = json["type"].asInt();
if (json.isMember("data")){
auto dataStr = json["data"].asString();
dataStr = base64_decode(dataStr);
data = std::vector<unsigned char>(dataStr.begin(), dataStr.end());
}
if (json.isMember("utype"))
user_type = json["utype"].asString();
const auto& jto = json[VALUE_KEY_TO];
if (jto.isString())
recipient = InfoHash(jto.asString());
const auto& jtype = json[VALUE_KEY_TYPE];
if (!jtype.isNull())
type = jtype.asInt();
const auto& jdata = json[VALUE_KEY_DATA];
if (jdata.isString())
data = base64_decode(jdata.asString());
const auto& jutype = json[VALUE_KEY_USERTYPE];
if (jutype.isString())
user_type = jutype.asString();
const auto& jprio = json["prio"];
if (jprio.isIntegral())
priority = jprio.asUInt();
}
Json::Value
Value::toJson() const
{
Json::Value val;
val["id"] = std::to_string(id);
val[VALUE_KEY_ID] = std::to_string(id);
if (isEncrypted()) {
val["cypher"] = base64_encode(cypher);
} else {
if (isSigned())
val["sig"] = base64_encode(signature);
val[VALUE_KEY_SIGNATURE] = base64_encode(signature);
bool has_owner = owner && *owner;
if (has_owner) { // isSigned
val["seq"] = seq;
val["owner"] = owner->toString();
val[VALUE_KEY_SEQ] = seq;
val[VALUE_KEY_OWNER] = owner->toString();
if (recipient)
val["to"] = recipient.toString();
val[VALUE_KEY_TO] = recipient.toString();
}
val["type"] = type;
val["data"] = base64_encode(data);
val[VALUE_KEY_TYPE] = type;
val[VALUE_KEY_DATA] = base64_encode(data);
if (not user_type.empty())
val["utype"] = user_type;
val[VALUE_KEY_USERTYPE] = user_type;
}
if (priority)
val["prio"] = priority;
return val;
}
@@ -239,13 +235,11 @@ uint64_t
unpackId(const Json::Value& json, const std::string& key) {
uint64_t ret = 0;
try {
if (json.isMember(key)) {
const auto& t = json[key];
if (t.isString()) {
ret = std::stoull(t.asString());
} else {
ret = t.asLargestUInt();
}
const auto& t = json[key];
if (t.isString()) {
ret = std::stoull(t.asString());
} else {
ret = t.asLargestUInt();
}
} catch (...) {}
return ret;
@@ -415,13 +409,13 @@ Select::Select(const std::string& q_str) {
while (std::getline(fields, token, ',')) {
trim_str(token);
if (token == "id")
if (token == VALUE_KEY_ID)
field(Value::Field::Id);
else if (token == "value_type")
field(Value::Field::ValueType);
else if (token == "owner_pk")
field(Value::Field::OwnerPk);
if (token == "seq")
if (token == VALUE_KEY_SEQ)
field(Value::Field::SeqNum);
else if (token == "user_type")
field(Value::Field::UserType);
@@ -457,13 +451,13 @@ Where::Where(const std::string& q_str) {
s = value_str.substr(1, value_str.size()-2);
else
s = value_str;
if (field_str == "id")
if (field_str == VALUE_KEY_ID)
id(v);
else if (field_str == "value_type")
valueType(v);
else if (field_str == "owner_pk")
owner(InfoHash(s));
else if (field_str == "seq")
else if (field_str == VALUE_KEY_SEQ)
seq(v);
else if (field_str == "user_type")
userType(s);
@@ -523,7 +517,7 @@ std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
for (auto fs = select.fieldSelection_.begin() ; fs != select.fieldSelection_.end() ; ++fs) {
switch (*fs) {
case Value::Field::Id:
s << "id";
s << VALUE_KEY_ID;
break;
case Value::Field::ValueType:
s << "value_type";
@@ -535,7 +529,7 @@ std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
s << "owner_public_key";
break;
case Value::Field::SeqNum:
s << "seq";
s << VALUE_KEY_SEQ;
break;
default:
break;
-1
View File
@@ -1 +0,0 @@
g++ -Wall -Wextra -Wfatal-errors -pedantic -g ../src/http.cpp tests_runner.cpp httptester.cpp -o run.out -lopendht -lpthread -lcppunit -ljsoncpp -DOPENDHT_JSONCPP -I../include/opendht -DASIO_STANDALONE
+71 -36
View File
@@ -19,41 +19,35 @@
#include "httptester.h"
// std
#include <opendht/value.h>
#include <opendht/dhtrunner.h>
#include <iostream>
#include <string>
#include <chrono>
#include <condition_variable>
namespace test {
CPPUNIT_TEST_SUITE_REGISTRATION(HttpTester);
void
HttpTester::setUp() {
logger = dht::log::getStdLogger();
nodePeer = std::make_shared<dht::DhtRunner>();
nodePeer->run(0);
nodePeer.run(0, /*identity*/{}, /*threaded*/true);
nodeProxy = std::make_shared<dht::DhtRunner>();
auto nodeProxy = std::make_shared<dht::DhtRunner>();
nodeProxy->run(0, /*identity*/{}, /*threaded*/true);
nodeProxy->bootstrap(nodePeer.getBound());
nodeProxy->bootstrap(nodePeer->getBound());
serverProxy = std::unique_ptr<dht::DhtProxyServer>(
new dht::DhtProxyServer(
/*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090", logger));
/*http*/dht::crypto::Identity{}, nodeProxy, 8080, /*pushServer*/"127.0.0.1:8090"));
}
void
HttpTester::tearDown() {
logger->d("[tester:http] stopping peer node");
nodePeer.join();
logger->d("[tester:http] stopping proxy server");
serverProxy.reset(nullptr);
logger->d("[tester:http] stopping proxy node");
nodeProxy->join();
serverProxy.reset();
nodePeer->join();
}
void
@@ -234,31 +228,72 @@ HttpTester::test_send_json() {
std::mutex cv_m;
std::unique_lock<std::mutex> lk(cv_m);
bool done = false;
dht::Value val {"hey"};
auto json = val.toJson();
json["permanent"] = false;
std::cout << "[test_send_json] sending:\n" << json << std::endl;
Json::Value resp_val;
unsigned int status = 0;
std::string url = "http://127.0.0.1:8080/key";
auto json = dht::Value("hey").toJson();
Json::Value resp_val;
// Act
auto request = std::make_shared<dht::http::Request>(serverProxy->io_context(), url, json,
[this, &cv, &done, &status, &resp_val](Json::Value value, unsigned int status_code){
if (status_code != 200 and logger)
logger->e("[tester] [status] failed with code=%i", status_code);
std::cout << "[tester] got response:\n" << value << std::endl;
resp_val = value;
status = status_code;
done = true;
cv.notify_all();
}, logger);
request->set_method(restinio::http_method_post());
auto request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"http://127.0.0.1:8080/key",
json,
[&](Json::Value value, unsigned int status_code) {
std::lock_guard<std::mutex> lk(cv_m);
resp_val = std::move(value);
status = status_code;
done = true;
cv.notify_all();
});
request->send();
// Assert
CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
CPPUNIT_ASSERT(status == 200);
CPPUNIT_ASSERT(resp_val["data"] == val.toJson()["data"]);
CPPUNIT_ASSERT_EQUAL(200u, status);
CPPUNIT_ASSERT_EQUAL(json["data"].asString(), resp_val["data"].asString());
done = false;
#if 0
request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"http://google.ca",
[&](const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
status = response.status_code;
done = true;
cv.notify_all();
});
request->send();
CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
CPPUNIT_ASSERT(status != 0);
done = false;
request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"https://google.ca",
[&](const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
status = response.status_code;
done = true;
cv.notify_all();
});
request->send();
CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
CPPUNIT_ASSERT(status != 0);
done = false;
request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"https://google.ca/sdbjklwGBIP",
[&](const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
status = response.status_code;
done = true;
cv.notify_all();
});
request->send();
CPPUNIT_ASSERT(cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; }));
CPPUNIT_ASSERT_EQUAL(404u, status);
#endif
}
} // namespace test
+5 -8
View File
@@ -23,12 +23,12 @@
#include <cppunit/TestFixture.h>
#include <cppunit/extensions/HelperMacros.h>
#include <opendht/value.h>
#include <opendht/log.h>
#include <opendht/http.h>
#include <opendht/dhtrunner.h>
#include <opendht/dht_proxy_server.h>
#include <asio.hpp>
#include <thread>
#include <memory>
namespace test {
@@ -88,10 +88,7 @@ class HttpTester : public CppUnit::TestFixture {
void test_send_json();
private:
std::shared_ptr<dht::Logger> logger {};
dht::DhtRunner nodePeer;
std::shared_ptr<dht::DhtRunner> nodeProxy;
std::shared_ptr<dht::DhtRunner> nodePeer;
std::unique_ptr<dht::DhtProxyServer> serverProxy;
};
+4 -1
View File
@@ -20,9 +20,12 @@
#include "peerdiscoverytester.h"
#include "opendht/value.h"
#include <mutex>
#include <condition_variable>
namespace test {
constexpr int MULTICAST_PORT = 2222;
constexpr unsigned MULTICAST_PORT = 2222;
const std::string DHT_NODE_NAME {"dht"};
const std::string JAMI_NODE_NAME {"jami"};
+3
View File
@@ -16,6 +16,9 @@
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "tools_common.h"
#include <opendht/rng.h>
+3
View File
@@ -18,6 +18,9 @@
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "tools_common.h"
extern "C" {
+3
View File
@@ -16,6 +16,9 @@
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "tools_common.h"
#include <opendht/node.h>
+7
View File
@@ -25,6 +25,13 @@
#include <opendht/log.h>
#include <opendht/crypto.h>
#include <opendht/network_utils.h>
#ifdef OPENDHT_INDEXATION
#include <opendht/indexation/pht.h>
#endif
#ifdef OPENDHT_PROXY_SERVER
#include <opendht/dht_proxy_server.h>
#endif
#ifndef WIN32_NATIVE
#include <getopt.h>
#include <readline/readline.h>