diff --git a/.ci/.env b/.ci/.env
index d17091b..0159808 100644
--- a/.ci/.env
+++ b/.ci/.env
@@ -6,6 +6,7 @@ LINUX_PACKAGES="make \
curl \
git \
libtool \
+ nasm \
ninja-build \
pkg-config \
python3.12 \
@@ -22,7 +23,9 @@ MACOS_PACKAGES="make \
rust \
curl \
git \
+ go \
libtool \
+ nasm \
ninja \
pkg-config \
python@3.12 \
diff --git a/BUILD.md b/BUILD.md
new file mode 100644
index 0000000..4d52618
--- /dev/null
+++ b/BUILD.md
@@ -0,0 +1,7 @@
+
+```bash
+brew install nasm # vcpkg liblsquic
+```
+
+`pip` can't install packages on MacOS.
+Specify venv directory with `-D Python3_ROOT_DIR=$PWD/.venv` cmake flag.
diff --git a/CMakeLists.txt b/CMakeLists.txt
index a84f22b..d571c12 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -11,6 +11,8 @@ if (TESTING)
list(APPEND VCPKG_MANIFEST_FEATURES test)
endif ()
+option(BUILD_EXAMPLES "Build examples" ON)
+
set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_CXX_EXTENSIONS OFF)
@@ -30,21 +32,22 @@ find_package(PkgConfig REQUIRED)
pkg_check_modules(libb2 REQUIRED IMPORTED_TARGET GLOBAL libb2)
find_package(Boost CONFIG REQUIRED COMPONENTS algorithm outcome program_options)
+find_package(Boost.DI CONFIG REQUIRED)
find_package(fmt CONFIG REQUIRED)
-find_package(yaml-cpp CONFIG REQUIRED)
find_package(jam_crust CONFIG REQUIRED)
+find_package(lsquic CONFIG REQUIRED)
+find_package(OpenSSL REQUIRED)
+find_package(prometheus-cpp CONFIG REQUIRED)
+find_package(qtils CONFIG REQUIRED)
find_package(scale CONFIG REQUIRED)
-find_package(soralog CONFIG REQUIRED)
find_package(schnorrkel_crust CONFIG REQUIRED)
-find_package(Boost.DI CONFIG REQUIRED)
-find_package(qtils CONFIG REQUIRED)
-find_package(prometheus-cpp CONFIG REQUIRED)
+find_package(soralog CONFIG REQUIRED)
+find_package(yaml-cpp CONFIG REQUIRED)
+find_package(ZLIB REQUIRED)
-add_library(headers INTERFACE)
-target_include_directories(headers INTERFACE
- $<BUILD_INTERFACE:${PROJECT_SOURCE_DIR}/src_>
- $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
-)
+include(vcpkg-overlay/cppcodec.cmake)
+
+include_directories(${CMAKE_SOURCE_DIR}/src)
add_subdirectory(src)
@@ -57,3 +60,7 @@ if (TESTING)
add_subdirectory(test-vectors)
add_subdirectory(tests)
endif ()
+
+if (BUILD_EXAMPLES)
+ add_subdirectory(example)
+endif ()
diff --git a/example/CMakeLists.txt b/example/CMakeLists.txt
new file mode 100644
index 0000000..efd41af
--- /dev/null
+++ b/example/CMakeLists.txt
@@ -0,0 +1,7 @@
+#
+# Copyright Quadrivium LLC
+# All Rights Reserved
+# SPDX-License-Identifier: Apache-2.0
+#
+
+add_subdirectory(snp_chat)
diff --git a/example/config.yaml b/example/config.yaml
index 1992f97..091782d 100644
--- a/example/config.yaml
+++ b/example/config.yaml
@@ -22,8 +22,9 @@ logging:
children:
- name: jam
children:
- - name: injector
- name: application
- - name: rpc
+ - name: injector
- name: metrics
- - name: threads
\ No newline at end of file
+ - name: rpc
+ - name: snp
+ - name: threads
diff --git a/example/snp_chat/CMakeLists.txt b/example/snp_chat/CMakeLists.txt
new file mode 100644
index 0000000..3c8c9e2
--- /dev/null
+++ b/example/snp_chat/CMakeLists.txt
@@ -0,0 +1,14 @@
+#
+# Copyright Quadrivium LLC
+# All Rights Reserved
+# SPDX-License-Identifier: Apache-2.0
+#
+
+add_executable(example_snp_chat
+ main.cpp
+)
+target_link_libraries(example_snp_chat
+ logger
+ snp
+)
+
diff --git a/example/snp_chat/main.cpp b/example/snp_chat/main.cpp
new file mode 100644
index 0000000..8187c00
--- /dev/null
+++ b/example/snp_chat/main.cpp
@@ -0,0 +1,244 @@
+#include <TODO_qtils/asio_buffer.hpp>
+#include <boost/asio/as_tuple.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/posix/stream_descriptor.hpp>
+#include <boost/asio/read_until.hpp>
+#include <boost/asio/streambuf.hpp>
+#include <boost/outcome/try.hpp>
+#include <fmt/format.h>
+#include <qtils/append.hpp>
+#include <qtils/unhex.hpp>
+
+#include "coro/spawn.hpp"
+#include "log/simple.hpp"
+#include "snp/connections/address.hpp"
+#include "snp/connections/connection.hpp"
+#include "snp/connections/connections.hpp"
+#include "snp/connections/controller.hpp"
+#include "snp/connections/stream.hpp"
+
+using jam::Coro;
+using jam::coroHandler;
+using jam::CoroHandler;
+using jam::CoroOutcome;
+using jam::coroSpawn;
+using jam::GenesisHash;
+using jam::IoContextPtr;
+using jam::crypto::ed25519::KeyPair;
+using jam::snp::Address;
+using jam::snp::ConnectionInfo;
+using jam::snp::Connections;
+using jam::snp::ConnectionsConfig;
+using jam::snp::ConnectionsController;
+using jam::snp::Key;
+using jam::snp::ProtocolId;
+using jam::snp::StreamPtr;
+
+auto logsys = jam::log::simpleLoggingSystem();
+
+inline auto operator""_ed25519(const char *c, size_t s) {
+ auto seed = qtils::unhex<jam::crypto::ed25519::Seed>({c, s}).value();
+ return jam::crypto::ed25519::from_seed(seed);
+}
+
+std::vector<KeyPair> keys{
+ "f8dfdb0f1103d9fb2905204ac32529d5f148761c4321b2865b0a40e15be75f57"_ed25519,
+ "96c891b8726cb18c781aefc082dbafcb827e16c8f18f22d461e83eabd618e780"_ed25519,
+ "619d5e68139f714ee8e7892ce5afd8fbe7a4172a675fea5c5a06fb94fe3d797d"_ed25519,
+ "8d0c5f498a763eaa8c04861cac06289784140b4bbfa814fef898f1f4095de4a3"_ed25519,
+};
+Address server_address{
+ Address::kLocal,
+ 10000,
+ jam::crypto::ed25519::get_public(keys[0]),
+};
+ProtocolId protocol_id = ProtocolId::make(0, true).value();
+
+size_t indexOfKey(const Key &key) {
+ auto it = std::ranges::find_if(keys, [&](const KeyPair &keypair) {
+ return jam::crypto::ed25519::get_public(keypair) == key;
+ });
+ if (it == keys.end()) {
+ throw std::logic_error{"TODO: example"};
+ }
+ return it - keys.begin();
+}
+
+struct ChatController : ConnectionsController {
+ static constexpr size_t kMaxMsg = 8;
+
+ struct Writer {
+ StreamPtr stream;
+ std::deque<qtils::Bytes> queue;
+ bool writing = false;
+ };
+ using WriterPtr = std::shared_ptr<Writer>;
+
+ std::map<size_t, WriterPtr> writers;
+
+ static CoroOutcome<void> write(WriterPtr writer,
+ size_t i_msg,
+ const std::string msg) {
+ qtils::Bytes buffer;
+ buffer.emplace_back(i_msg);
+ qtils::append(buffer, qtils::str2byte(msg));
+ writer->queue.emplace_back(buffer);
+ if (writer->writing) {
+ co_return outcome::success();
+ }
+ writer->writing = true;
+ while (not writer->queue.empty()) {
+ auto buffer = writer->queue.front();
+ writer->queue.pop_front();
+ BOOST_OUTCOME_CO_TRY(
+ co_await writer->stream->write(writer->stream, buffer));
+ }
+ writer->writing = false;
+ co_return outcome::success();
+ }
+
+ void onOpen(Key key) override {
+ fmt::println("#{} (connected)", indexOfKey(key));
+ }
+
+ void onClose(Key key) override {
+ fmt::println("#{} (disconnected)", indexOfKey(key));
+ }
+
+ void print(size_t i_msg, std::string msg) {
+ fmt::println("#{} > {}", i_msg, msg);
+ }
+
+ Coro<void> broadcast(std::optional<size_t> i_read,
+ size_t i_msg,
+ std::string msg) {
+ for (auto &[i_write, writer] : writers) {
+ if (i_write == i_read) {
+ continue;
+ }
+ co_await coroSpawn([this, i_write, writer, i_msg, msg]() -> Coro<void> {
+ if (not co_await write(writer, i_msg, msg)) {
+ writers.erase(i_write);
+ }
+ });
+ }
+ }
+
+ Coro<void> onRead(size_t i_read, size_t i_msg, std::string msg) {
+ print(i_msg, msg);
+ co_await broadcast(i_read, i_msg, msg);
+ }
+
+ CoroOutcome<void> add(ConnectionInfo info, StreamPtr stream) {
+ auto i_read = indexOfKey(info.key);
+ writers.emplace(i_read, std::make_shared<Writer>(Writer{stream}));
+ qtils::Bytes buffer;
+ while (true) {
+ BOOST_OUTCOME_CO_TRY(auto read,
+ co_await stream->read(stream, buffer, 1 + kMaxMsg));
+ if (not read) {
+ break;
+ }
+ if (buffer.size() < 1) {
+ break;
+ }
+ auto i_msg = buffer[0];
+ co_await onRead(
+ i_read, i_msg, std::string{qtils::byte2str(buffer).substr(1)});
+ }
+ co_await stream->shutdownRead(stream);
+ co_return outcome::success();
+ }
+};
+
+struct Input {
+ Input(IoContextPtr io_context_ptr) : fd_{*io_context_ptr, STDIN_FILENO} {}
+
+ Coro<std::optional<std::string>> read() {
+ auto [ec, n] = co_await boost::asio::async_read_until(
+ fd_, buf_, "\n", boost::asio::as_tuple(boost::asio::use_awaitable));
+ if (ec) {
+ co_return std::nullopt;
+ }
+ auto s = qtils::byte2str(qtils::asioBuffer(buf_.data()));
+ auto i = s.find("\n");
+ if (i != s.npos) {
+ s = s.substr(0, i);
+ }
+ auto r = std::string{s};
+ buf_.consume(buf_.size());
+ co_return r;
+ }
+
+ boost::asio::posix::stream_descriptor fd_;
+ boost::asio::streambuf buf_;
+};
+
+CoroOutcome<void> co_main(IoContextPtr io_context_ptr, size_t arg_i) {
+ fmt::println("#{} (self)", arg_i);
+
+ std::optional<uint16_t> listen_port;
+ GenesisHash genesis;
+ ConnectionsConfig config{genesis, keys.at(arg_i)};
+ auto is_server = arg_i == 0;
+ if (is_server) {
+ config.listen_port = server_address.port;
+ }
+ auto connections =
+ std::make_shared<Connections>(io_context_ptr, logsys, config);
+ auto chat = std::make_shared<ChatController>();
+ BOOST_OUTCOME_CO_TRY(co_await connections->init(connections, chat));
+ co_await coroSpawn([io_context_ptr, arg_i, chat]() -> Coro<void> {
+ Input input{io_context_ptr};
+ while (true) {
+ auto msg = co_await input.read();
+ if (not msg.has_value()) {
+ break;
+ }
+ msg->resize(std::min(msg->size(), ChatController::kMaxMsg));
+ if (msg->empty()) {
+ continue;
+ }
+ co_await chat->broadcast(std::nullopt, arg_i, *msg);
+ }
+ io_context_ptr->stop();
+ });
+ if (not is_server) {
+ BOOST_OUTCOME_CO_TRY(
+ auto connection,
+ co_await connections->connect(connections, server_address));
+ BOOST_OUTCOME_CO_TRY(auto stream,
+ co_await connection->open(connection, protocol_id));
+ std::ignore = co_await chat->add(connection->info(), stream);
+ fmt::println("(disconnected)");
+ io_context_ptr->stop();
+ } else {
+ co_await connections->serve(
+ connections,
+ protocol_id,
+ [chat](ConnectionInfo info, StreamPtr stream) -> CoroOutcome<void> {
+ co_return co_await chat->add(info, stream);
+ });
+ std::optional<CoroHandler<void>> work_guard;
+ co_await coroHandler<void>([&](CoroHandler<void> &&handler) {
+ work_guard.emplace(std::move(handler));
+ });
+ }
+ co_return outcome::success();
+}
+
+int main(int argc, char **argv) {
+ setvbuf(stdout, nullptr, _IONBF, 0);
+ setvbuf(stderr, nullptr, _IONBF, 0);
+
+ size_t arg_i = 0;
+ if (argc == 2) {
+ arg_i = std::atoi(argv[1]);
+ }
+
+ auto io_context_ptr = std::make_shared<boost::asio::io_context>();
+ coroSpawn(*io_context_ptr, [io_context_ptr, arg_i]() -> Coro<void> {
+ (co_await co_main(io_context_ptr, arg_i)).value();
+ });
+ io_context_ptr->run();
+}
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 08f0356..36888c0 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -4,8 +4,6 @@
# SPDX-License-Identifier: Apache-2.0
#
-include_directories(${CMAKE_CURRENT_SOURCE_DIR})
-
# Executables (should contain `main()` function)
add_subdirectory(executable)
@@ -24,3 +22,6 @@ add_subdirectory(metrics)
# Clocks and time subsystem
add_subdirectory(clock)
+# Simple Network Protocol
+add_subdirectory(snp)
+
diff --git a/src/TODO_qtils/asio_buffer.hpp b/src/TODO_qtils/asio_buffer.hpp
new file mode 100644
index 0000000..f613833
--- /dev/null
+++ b/src/TODO_qtils/asio_buffer.hpp
@@ -0,0 +1,31 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <boost/asio/buffer.hpp>
+#include <qtils/bytes.hpp>
+
+namespace qtils {
+ inline boost::asio::const_buffer asioBuffer(BytesIn s) {
+ return {s.data(), s.size()};
+ }
+
+ boost::asio::mutable_buffer asioBuffer(auto &&t)
+ requires(requires { BytesOut{t}; })
+ {
+ BytesOut s{t};
+ return {s.data(), s.size()};
+ }
+
+ inline BytesIn asioBuffer(const boost::asio::const_buffer &s) {
+ return {static_cast<const uint8_t *>(s.data()), s.size()};
+ }
+
+ inline BytesOut asioBuffer(const boost::asio::mutable_buffer &s) {
+ return {static_cast<uint8_t *>(s.data()), s.size()};
+ }
+} // namespace qtils
diff --git a/src/TODO_qtils/from_span.hpp b/src/TODO_qtils/from_span.hpp
new file mode 100644
index 0000000..0a5636a
--- /dev/null
+++ b/src/TODO_qtils/from_span.hpp
@@ -0,0 +1,32 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstring>
+#include <optional>
+#include <stdexcept>
+
+#include <qtils/bytes.hpp>
+
+namespace qtils {
+ inline bool fromSpan(BytesOut out, BytesIn span) {
+ if (span.size() != out.size()) {
+ return false;
+ }
+ memcpy(out.data(), span.data(), out.size());
+ return true;
+ }
+
+ template <typename T>
+ std::optional<T> fromSpan(BytesIn span) {
+ T out;
+ if (not fromSpan(out, span)) {
+ return std::nullopt;
+ }
+ return out;
+ }
+} // namespace qtils
diff --git a/src/TODO_qtils/make_shared_private.hpp b/src/TODO_qtils/make_shared_private.hpp
new file mode 100644
index 0000000..6405c38
--- /dev/null
+++ b/src/TODO_qtils/make_shared_private.hpp
@@ -0,0 +1,36 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace qtils {
+ /**
+ * `enable_shared_from_this` requires `make_shared`,
+ * which requires public constructor.
+ * `injector` may accidentially call wrong constructor.
+ * Using `MakeSharedPrivate` argument in public constructor prevents injector
+ * from using it.
+ * class Foo : public std::enable_shared_from_this<Foo> {
+ * public:
+ * Foo(MakeSharedPrivate, ...);
+ * static auto factory(...) {
+ * return MakeSharedPrivate::make(...);
+ * }
+ * };
+ */
+ class MakeSharedPrivate {
+ MakeSharedPrivate() = default;
+
+ public:
+ template <typename T, typename... A>
+ static std::shared_ptr<T> make(A &&...args) {
+ return std::make_shared<T>(MakeSharedPrivate{},
+ std::forward<decltype(args)>(args)...);
+ }
+ };
+} // namespace qtils
diff --git a/src/TODO_qtils/map_entry.hpp b/src/TODO_qtils/map_entry.hpp
new file mode 100644
index 0000000..b616fef
--- /dev/null
+++ b/src/TODO_qtils/map_entry.hpp
@@ -0,0 +1,96 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <map>
+#include <stdexcept>
+#include <unordered_map>
+#include <variant>
+
+namespace qtils {
+ template <typename M>
+ struct MapEntry {
+ using I = typename M::iterator;
+ using K = typename M::key_type;
+
+ MapEntry(M &map, const K &key) : map{map} {
+ if (auto it = map.find(key); it != map.end()) {
+ it_or_key = it;
+ } else {
+ it_or_key = key;
+ }
+ }
+
+ bool has() const {
+ return std::holds_alternative<I>(it_or_key);
+ }
+ operator bool() const {
+ return has();
+ }
+ auto &operator*() {
+ if (not has()) {
+ throw std::logic_error{
+ "Call dereference operator of MapEntry without valid iterator"};
+ }
+ return std::get<I>(it_or_key)->second;
+ }
+ auto *operator->() {
+ if (not has()) {
+ throw std::logic_error{
+ "Call member access through pointer operator of MapEntry without "
+ "valid iterator"};
+ }
+ return &std::get<I>(it_or_key)->second;
+ }
+ void insert(M::mapped_type value) {
+ if (has()) {
+ throw std::logic_error{"MapEntry::insert"};
+ }
+ it_or_key =
+ map.emplace(std::move(std::get<K>(it_or_key)), std::move(value))
+ .first;
+ }
+ void insert_or_assign(M::mapped_type value) {
+ if (not has()) {
+ insert(std::move(value));
+ } else {
+ **this = std::move(value);
+ }
+ }
+ /// Remove from map and return value.
+ [[nodiscard]] M::mapped_type extract() {
+ if (not has()) {
+ throw std::logic_error{"MapEntry::extract"};
+ }
+ auto node = map.extract(std::get<I>(it_or_key));
+ it_or_key = std::move(node.key());
+ return std::move(node.mapped());
+ }
+
+ void erase() {
+ if (has()) {
+ auto it = std::get<I>(it_or_key);
+ it_or_key = it->first;
+ map.erase(it);
+ }
+ }
+
+ // NOLINTNEXTLINE(cppcoreguidelines-avoid-const-or-ref-data-members)
+ M ↦
+ std::variant<I, K> it_or_key{};
+ };
+
+ template <typename K, typename V, typename L, typename A>
+ auto entry(std::map<K, V, L, A> &map, const K &key) {
+ return MapEntry<std::remove_cvref_t<decltype(map)>>{map, key};
+ }
+
+ template <typename K, typename V, typename H, typename E, typename A>
+ auto entry(std::unordered_map<K, V, H, E, A> &map, const K &key) {
+ return MapEntry<std::remove_cvref_t<decltype(map)>>{map, key};
+ }
+} // namespace qtils
diff --git a/src/TODO_qtils/std_hash_of.hpp b/src/TODO_qtils/std_hash_of.hpp
new file mode 100644
index 0000000..b1fe861
--- /dev/null
+++ b/src/TODO_qtils/std_hash_of.hpp
@@ -0,0 +1,17 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <utility>
+
+namespace qtils {
+ template <typename T>
+ requires(requires(const T &v) { std::hash<T>()(v); })
+ size_t stdHashOf(const T &v) {
+ return std::hash<T>()(v);
+ }
+} // namespace qtils
diff --git a/src/TODO_qtils/variant_get.hpp b/src/TODO_qtils/variant_get.hpp
new file mode 100644
index 0000000..84cecf1
--- /dev/null
+++ b/src/TODO_qtils/variant_get.hpp
@@ -0,0 +1,41 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <variant>
+
+namespace qtils {
+ /**
+ * Some people don't like pointer in
+ * if (auto *t = std::get_if<T>(&v))
+ */
+ template <typename T, typename V>
+ struct VariantGet {
+ VariantGet(V &variant) : variant_{variant} {}
+ operator bool() const {
+ return std::holds_alternative<T>(variant_);
+ }
+ auto &operator*() {
+ return std::get<T>(variant_);
+ }
+ auto *operator->() {
+ return &std::get<T>(variant_);
+ }
+ // NOLINTNEXTLINE(cppcoreguidelines-avoid-const-or-ref-data-members)
+ V &variant_;
+ };
+
+ template <typename T, typename... V>
+ auto variantGet(const std::variant<V...> &variant) {
+ return VariantGet<T, decltype(variant)>{variant};
+ }
+
+ template <typename T, typename... V>
+ auto variantGet(std::variant<V...> &variant) {
+ return VariantGet<T, decltype(variant)>{variant};
+ }
+} // namespace qtils
diff --git a/src/coro/coro.hpp b/src/coro/coro.hpp
new file mode 100644
index 0000000..99c29a7
--- /dev/null
+++ b/src/coro/coro.hpp
@@ -0,0 +1,49 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <boost/asio/awaitable.hpp>
+#include <qtils/outcome.hpp>
+
+namespace jam {
+ /**
+ * Return type for coroutine.
+ *
+ * Does not resume when:
+ * - called directly outside executor, returns coroutine.
+ * - `coroSpawn` called when not running inside executor,
+ * resumes on next executor tick.
+ * int main() {
+ * boost::asio::io_context io;
+ * coroSpawn(io, []() -> Coro<void> { co_return; }); // suspended
+ * io.run_one(); // resumes
+ * // may complete before next statement
+ * }
+ * Resumes when:
+ * - `coroSpawn` when running inside specified executor.
+ * post(executor, [] {
+ * coroSpawn(executor, []() -> Coro<void> { co_return; }) // resumes
+ * // may complete before next statement
+ * })
+ * co_await coroSpawn([]() -> Coro<void> { co_return; }) // resumes
+ * // may complete before next statement
+ * - `co_await`
+ * co_await foo() // resumes
+ * // may complete before next statement
+ * After resuming may complete before specified statement ends.
+ *
+ * Use `CORO_YIELD` explicitly to suspend coroutine until next executor tick.
+ */
+ template <typename T>
+ using Coro = boost::asio::awaitable<T>;
+
+ /**
+ * Return type for coroutine returning outcome.
+ */
+ template <typename T>
+ using CoroOutcome = Coro<outcome::result<T>>;
+} // namespace jam
diff --git a/src/coro/future.hpp b/src/coro/future.hpp
new file mode 100644
index 0000000..9dad641
--- /dev/null
+++ b/src/coro/future.hpp
@@ -0,0 +1,69 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <deque>
+#include <stdexcept>
+#include <variant>
+
+#include <TODO_qtils/variant_get.hpp>
+
+#include "coro/handler.hpp"
+#include "coro/set_thread.hpp"
+
+namespace jam {
+ template <typename T>
+ class SharedFuture {
+ public:
+ using SelfSPtr = std::shared_ptr<SharedFuture<T>>;
+
+ SharedFuture(IoContextPtr io_context_ptr)
+ : io_context_ptr_{std::move(io_context_ptr)} {}
+
+ static Coro<bool> ready(SelfSPtr self) {
+ co_await setCoroThread(self->io_context_ptr_);
+ co_return std::holds_alternative<T>(self->state_);
+ }
+
+ /**
+ * Resumes coroutine immediately or inside `set`.
+ */
+ static Coro<T> get(SelfSPtr self) {
+ co_await setCoroThread(self->io_context_ptr_);
+ if (auto value = qtils::variantGet<T>(self->state_)) {
+ co_return *value;
+ }
+ auto &handlers = std::get<Handlers>(self->state_);
+ co_return co_await coroHandler<T>([&](CoroHandler<T> &&handler) {
+ handlers.emplace_back(std::move(handler));
+ });
+ }
+
+ /**
+ * Set value and wake waiting coroutines.
+ * Coroutines may complete before `set` returns.
+ */
+ static Coro<void> set(SelfSPtr self, T value) {
+ co_await setCoroThread(self->io_context_ptr_);
+ if (std::holds_alternative<T>(self->state_)) {
+ throw std::logic_error{"SharedFuture::set must be called once"};
+ }
+ auto handlers = std::move(std::get<Handlers>(self->state_));
+ self->state_ = std::move(value);
+ auto &state_value = std::get<T>(self->state_);
+ for (auto &handler : handlers) {
+ handler(state_value);
+ }
+ }
+
+ private:
+ using Handlers = std::deque<CoroHandler<T>>;
+
+ IoContextPtr io_context_ptr_;
+ std::variant<Handlers, T> state_;
+ };
+} // namespace jam
diff --git a/src/coro/handler.hpp b/src/coro/handler.hpp
new file mode 100644
index 0000000..696175e
--- /dev/null
+++ b/src/coro/handler.hpp
@@ -0,0 +1,35 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <concepts>
+
+#include <boost/asio/use_awaitable.hpp>
+
+#include "coro/coro.hpp"
+
+namespace jam {
+ template <typename T>
+ using CoroHandler = std::conditional_t<
+ std::is_void_v<T>,
+ boost::asio::detail::awaitable_handler<typename Coro<T>::executor_type>,
+ boost::asio::detail::awaitable_handler<typename Coro<T>::executor_type,
+ T>>;
+
+ /**
+ * Create handler for coroutine.
+ * Coroutine may complete earlier than handler returns.
+ */
+ template <typename T>
+ Coro<T> coroHandler(std::invocable<CoroHandler<T> &&> auto &&f) {
+ co_await [&](auto *frame) {
+ f(CoroHandler<T>{frame->detach_thread()});
+ return nullptr;
+ };
+ abort();
+ }
+} // namespace jam
diff --git a/src/coro/init.hpp b/src/coro/init.hpp
new file mode 100644
index 0000000..834cf61
--- /dev/null
+++ b/src/coro/init.hpp
@@ -0,0 +1,75 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "coro/future.hpp"
+#include "coro/spawn.hpp"
+
+namespace jam {
+ /**
+ * Async init flag.
+ * struct Foo {
+ * CoroOutcome<void> init() {
+ * auto init = init_.init(); // dtor will fail incomplete init
+ * ...
+ * init.ready(); // init completed
+ * }
+ * CoroOutcome<void> foo() {
+ * if (not co_await init_.ready()) // init failed
+ * ... // ready
+ * }
+ * CoroInit init_;
+ * }
+ */
+ class CoroInit {
+ class Init {
+ public:
+ Init(CoroInit &self) : self_{self} {}
+ ~Init() {
+ self_.set(false);
+ }
+ void ready() {
+ self_.set(true);
+ }
+
+ private:
+ CoroInit &self_;
+ };
+
+ public:
+ CoroInit(IoContextPtr io_context_ptr)
+ : io_context_ptr_{std::move(io_context_ptr)},
+ future_{std::make_shared<decltype(future_)::element_type>(
+ io_context_ptr_)} {}
+
+ auto init() {
+ if (init_called_) {
+ throw std::logic_error{"Coro::init init must be called once"};
+ }
+ init_called_ = true;
+ return Init{*this};
+ }
+
+ Coro<bool> ready() {
+ return future_->get(future_);
+ }
+
+ private:
+ void set(bool ready) {
+ coroSpawn(*io_context_ptr_, [future{future_}, ready]() -> Coro<void> {
+ if (not ready and co_await future->ready(future)) {
+ co_return;
+ }
+ co_await future->set(future, ready);
+ });
+ }
+
+ IoContextPtr io_context_ptr_;
+ std::shared_ptr<SharedFuture<bool>> future_;
+ bool init_called_ = false;
+ };
+} // namespace jam
diff --git a/src/coro/io_context_ptr.hpp b/src/coro/io_context_ptr.hpp
new file mode 100644
index 0000000..d88248e
--- /dev/null
+++ b/src/coro/io_context_ptr.hpp
@@ -0,0 +1,17 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace boost::asio {
+ class io_context;
+} // namespace boost::asio
+
+namespace jam {
+ using IoContextPtr = std::shared_ptr<boost::asio::io_context>;
+} // namespace jam
diff --git a/src/coro/set_thread.hpp b/src/coro/set_thread.hpp
new file mode 100644
index 0000000..f84e401
--- /dev/null
+++ b/src/coro/set_thread.hpp
@@ -0,0 +1,22 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/use_awaitable.hpp>
+
+#include "coro/coro.hpp"
+#include "coro/io_context_ptr.hpp"
+
+namespace jam {
+ inline Coro<void> setCoroThread(IoContextPtr io_context_ptr) {
+ if (not io_context_ptr->get_executor().running_in_this_thread()) {
+ co_await boost::asio::post(*io_context_ptr, boost::asio::use_awaitable);
+ }
+ }
+} // namespace jam
diff --git a/src/coro/spawn.hpp b/src/coro/spawn.hpp
new file mode 100644
index 0000000..8829acd
--- /dev/null
+++ b/src/coro/spawn.hpp
@@ -0,0 +1,68 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <tuple>
+
+#include <boost/asio/co_spawn.hpp>
+
+#include "coro/coro.hpp"
+
+namespace jam {
+ template <typename T>
+ concept CoroSpawnExecutor =
+ boost::asio::is_executor<T>::value
+ || boost::asio::execution::is_executor<T>::value
+ || std::is_convertible_v<T, boost::asio::execution_context &>;
+
+ void coroSpawn(CoroSpawnExecutor auto &&executor, Coro<void> &&coro) {
+ boost::asio::co_spawn(std::forward<decltype(executor)>(executor),
+ std::move(coro),
+ [](std::exception_ptr e) {
+ if (e != nullptr) {
+ std::rethrow_exception(e);
+ }
+ });
+ }
+
+ template <typename T>
+ void coroSpawn(CoroSpawnExecutor auto &&executor, Coro<T> &&coro) {
+ coroSpawn(std::forward<decltype(executor)>(executor),
+ [coro{std::move(coro)}]() mutable -> Coro<void> {
+ std::ignore = co_await std::move(coro);
+ });
+ }
+
+ /**
+ * Start coroutine on specified executor.
+ * Spawning on same executor would execute coroutine immediately,
+ * so coroutine may complete before `coroSpawn` returns.
+ * Prevents dangling lambda capture in `coroSpawn([capture] { ... })`.
+ * `co_spawn([capture] { ... })` doesn't work
+ * because lambda is destroyed after returning coroutine object.
+ * `co_spawn([](args){ ... }(capture))`
+ * works because arguments are stored in coroutine state.
+ */
+ void coroSpawn(CoroSpawnExecutor auto &&executor, auto &&f) {
+ coroSpawn(std::forward<decltype(executor)>(executor),
+ [](std::remove_cvref_t<decltype(f)> f) -> Coro<void> {
+ if constexpr (std::is_void_v<decltype(f().await_resume())>) {
+ co_await f();
+ } else {
+ std::ignore = co_await f();
+ }
+ }(std::forward<decltype(f)>(f)));
+ }
+
+ /**
+ * `coroSpawn` with current coroutine executor.
+ */
+ Coro<void> coroSpawn(auto f) {
+ coroSpawn(co_await boost::asio::this_coro::executor, std::move(f));
+ co_return;
+ }
+} // namespace jam
diff --git a/src/coro/weak.hpp b/src/coro/weak.hpp
new file mode 100644
index 0000000..fc8e95d
--- /dev/null
+++ b/src/coro/weak.hpp
@@ -0,0 +1,50 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "coro/coro.hpp"
+
+/**
+ * Converts shared pointer to weak pointer for `co_await` duration.
+ * Shared pointer owner can cancel operation by destroying shared pointer.
+ * Checks that state wasn't destroyed during `co_await` before continuing.
+ * auto cb2 = [cb, tmp_weak = std::weak_ptr{shared}, ...](auto r) {
+ * auto shared = tmp_weak.lock();
+ * if (not shared) {
+ * return cb(...);
+ * }
+ * ...
+ * cb();
+ * };
+ */
+#define _CORO_WEAK_AWAIT(tmp_weak, tmp_coro, auto_r, r, shared, coro, ...) \
+ ({ \
+ auto tmp_weak = std::weak_ptr{shared}; \
+ /* coroutine constructor may need `shared` alive */ \
+ auto tmp_coro = (coro); \
+ /* reset `shared` after coroutine is constructed */ \
+ shared.reset(); \
+ auto_r co_await std::move(tmp_coro); \
+ shared = tmp_weak.lock(); \
+ if (not shared) { \
+ co_return __VA_ARGS__; \
+ } \
+ r \
+ })
+#define CORO_WEAK_AWAIT(shared, coro, ...) \
+ _CORO_WEAK_AWAIT( \
+ QTILS_UNIQUE_NAME(tmp_weak), QTILS_UNIQUE_NAME(tmp_coro), auto r =, r; \
+ , shared, coro, __VA_ARGS__)
+
+#define CORO_WEAK_AWAIT_V(shared, coro, ...) \
+ _CORO_WEAK_AWAIT(QTILS_UNIQUE_NAME(tmp_weak), \
+ QTILS_UNIQUE_NAME(tmp_coro), \
+ , \
+ , \
+ shared, \
+ coro, \
+ __VA_ARGS__)
diff --git a/src/coro/yield.hpp b/src/coro/yield.hpp
new file mode 100644
index 0000000..3e6b2a3
--- /dev/null
+++ b/src/coro/yield.hpp
@@ -0,0 +1,23 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <boost/asio/post.hpp>
+#include <boost/asio/this_coro.hpp>
+#include <boost/asio/use_awaitable.hpp>
+
+#include "coro/coro.hpp"
+
+namespace jam {
+ /**
+ * Thread switch operation always completes, so it can't leak `shared_ptr`.
+ */
+ Coro<void> coroYield() {
+ co_await boost::asio::post(co_await boost::asio::this_coro::executor,
+ boost::asio::use_awaitable);
+ }
+} // namespace jam
diff --git a/src/crypto/ed25519.hpp b/src/crypto/ed25519.hpp
index 224d46c..441bf9e 100644
--- a/src/crypto/ed25519.hpp
+++ b/src/crypto/ed25519.hpp
@@ -4,10 +4,15 @@
* SPDX-License-Identifier: Apache-2.0
*/
+#pragma once
+
#include <schnorrkel_crust.h>
+
+#include <TODO_qtils/from_span.hpp>
#include <qtils/bytes.hpp>
namespace jam::crypto::ed25519 {
+ using Seed = qtils::BytesN<ED25519_SEED_LENGTH>;
using Secret = qtils::BytesN<ED25519_SECRET_KEY_LENGTH>;
using Public = qtils::BytesN<ED25519_PUBLIC_KEY_LENGTH>;
using KeyPair = qtils::BytesN<ED25519_KEYPAIR_LENGTH>;
@@ -34,4 +39,22 @@ namespace jam::crypto::ed25519 {
message.size_bytes());
return res == ED25519_RESULT_OK;
}
-} // namespace jam::ed25519
+
+ inline KeyPair from_seed(const Seed &seed) {
+ KeyPair keypair;
+ ed25519_keypair_from_seed(keypair.data(), seed.data());
+ return keypair;
+ }
+
+ inline Public get_public(const KeyPair &keypair) {
+ return qtils::fromSpan<Public>(
+ std::span{keypair}.subspan(ED25519_SECRET_KEY_LENGTH))
+ .value();
+ }
+
+ inline Public get_secret(const KeyPair &keypair) {
+ return qtils::fromSpan<Secret>(
+ std::span{keypair}.first(ED25519_SECRET_KEY_LENGTH))
+ .value();
+ }
+} // namespace jam::crypto::ed25519
diff --git a/src/executable/CMakeLists.txt b/src/executable/CMakeLists.txt
index 392671f..742a8e8 100644
--- a/src/executable/CMakeLists.txt
+++ b/src/executable/CMakeLists.txt
@@ -11,8 +11,6 @@ set(LIBRARIES
)
include_directories(
- ${PROJECT_SOURCE_DIR}
- ${CMAKE_SOURCE_DIR}/src
${CMAKE_BINARY_DIR}/generated
)
diff --git a/src/injector/node_injector.cpp b/src/injector/node_injector.cpp
index 48ef7d5..26962a5 100644
--- a/src/injector/node_injector.cpp
+++ b/src/injector/node_injector.cpp
@@ -50,6 +50,7 @@ namespace {
di::bind<log::LoggingSystem>.to(logsys),
di::bind<metrics::Handler>.to<metrics::PrometheusHandler>(),
di::bind<metrics::Exposer>.to<metrics::ExposerImpl>(),
+ useConfig(metrics::Session::Configuration{}),
di::bind<metrics::Exposer::Configuration>.to([](const auto &injector) {
return metrics::Exposer::Configuration{
{boost::asio::ip::address_v4::from_string("127.0.0.1"), 7777}
@@ -92,7 +93,7 @@ namespace jam::injector {
NodeInjector::NodeInjector(std::shared_ptr<log::LoggingSystem> logsys,
std::shared_ptr<app::Configuration> config)
: pimpl_{std::make_unique<NodeInjectorImpl>(
- makeNodeInjector(std::move(logsys), std::move(config)))} {}
+ makeNodeInjector(std::move(logsys), std::move(config)))} {}
std::shared_ptr<app::Application> NodeInjector::injectApplication() {
return pimpl_->injector_
diff --git a/src/log/simple.hpp b/src/log/simple.hpp
new file mode 100644
index 0000000..0694f6a
--- /dev/null
+++ b/src/log/simple.hpp
@@ -0,0 +1,34 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <soralog/impl/configurator_from_yaml.hpp>
+
+#include "log/logger.hpp"
+
+namespace jam::log {
+ std::shared_ptr<LoggingSystem> simpleLoggingSystem() {
+ std::string yaml = R"(
+ sinks:
+ - name: console
+ type: console
+ capacity: 4
+ latency: 0
+ groups:
+ - name: main
+ sink: console
+ level: info
+ is_fallback: true
+ )";
+ auto logsys = std::make_shared<soralog::LoggingSystem>(
+ std::make_shared<soralog::ConfiguratorFromYAML>(yaml));
+ if (auto r = logsys->configure().message; not r.empty()) {
+ fmt::println(stderr, "soralog error: {}", r);
+ }
+ return std::make_shared<LoggingSystem>(logsys);
+ }
+} // namespace jam::log
diff --git a/src/snp/CMakeLists.txt b/src/snp/CMakeLists.txt
new file mode 100644
index 0000000..81b815a
--- /dev/null
+++ b/src/snp/CMakeLists.txt
@@ -0,0 +1,25 @@
+#
+# Copyright Quadrivium LLC
+# All Rights Reserved
+# SPDX-License-Identifier: Apache-2.0
+#
+
+add_library(snp
+ connections/alpn.cpp
+ connections/connection.cpp
+ connections/connections.cpp
+ connections/dns_name.cpp
+ connections/lsquic/engine.cpp
+ connections/lsquic/init.cpp
+ connections/prefer_key.cpp
+ connections/protocol_id.cpp
+ connections/stream.cpp
+ connections/tls_certificate.cpp
+)
+target_link_libraries(snp
+ fmt::fmt
+ lsquic::lsquic
+ OpenSSL::SSL
+ schnorrkel_crust::schnorrkel_crust
+ ZLIB::ZLIB
+)
diff --git a/src/snp/connections/address.hpp b/src/snp/connections/address.hpp
new file mode 100644
index 0000000..baecd87
--- /dev/null
+++ b/src/snp/connections/address.hpp
@@ -0,0 +1,20 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "snp/connections/key.hpp"
+
+namespace jam::snp {
+ struct Address {
+ using Ip = qtils::BytesN<16>;
+ static constexpr Ip kLocal{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1};
+
+ Ip ip;
+ uint16_t port;
+ Key key;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/alpn.cpp b/src/snp/connections/alpn.cpp
new file mode 100644
index 0000000..1fe4c56
--- /dev/null
+++ b/src/snp/connections/alpn.cpp
@@ -0,0 +1,47 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/alpn.hpp"
+
+#include <openssl/ssl.h>
+#include <qtils/append.hpp>
+#include <qtils/bytestr.hpp>
+#include <qtils/hex.hpp>
+
+#include "snp/connections/error.hpp"
+
+namespace jam::snp {
+ Alpn::Alpn(const GenesisHash &genesis) {
+ const auto protocol =
+ fmt::format("jamnp-s/{}/{:x}", kVersion, std::span{genesis}.first(4));
+ bytes_.reserve(1 + protocol.size());
+ bytes_.emplace_back(protocol.size());
+ qtils::append(bytes_, qtils::str2byte(protocol));
+ }
+
+ outcome::result<void> Alpn::set(ssl_ctx_st *ssl_ctx) {
+ if (SSL_CTX_set_alpn_protos(ssl_ctx, bytes_.data(), bytes_.size()) != 0) {
+ return OpenSslError::SSL_CTX_set_alpn_protos;
+ }
+ SSL_CTX_set_alpn_select_cb(ssl_ctx, select, this);
+ return outcome::success();
+ }
+
+ int Alpn::select(ssl_st *ssl,
+ const unsigned char **out,
+ unsigned char *outlen,
+ const unsigned char *in,
+ unsigned int inlen,
+ void *void_self) {
+ auto *self = static_cast<Alpn *>(void_self);
+ uint8_t *out2 = nullptr;
+ int r = SSL_select_next_proto(
+ &out2, outlen, in, inlen, self->bytes_.data(), self->bytes_.size());
+ *out = out2;
+ return r == OPENSSL_NPN_NEGOTIATED ? SSL_TLSEXT_ERR_OK
+ : SSL_TLSEXT_ERR_ALERT_FATAL;
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/alpn.hpp b/src/snp/connections/alpn.hpp
new file mode 100644
index 0000000..0df2988
--- /dev/null
+++ b/src/snp/connections/alpn.hpp
@@ -0,0 +1,50 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <string>
+
+#include <qtils/outcome.hpp>
+
+#include "types/genesis_hash.hpp"
+
+struct ssl_ctx_st;
+struct ssl_st;
+
+namespace jam::snp {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L30-L41
+ /**
+ * TLS ALPN (Application-Layer Protocol Negotiation) used by jam peers.
+ */
+ class Alpn {
+ static constexpr auto kVersion = 0;
+
+ public:
+ /**
+ * Make ALPN from jam genesis hash.
+ */
+ Alpn(const GenesisHash &genesis);
+
+ /**
+ * Set ALPN for `ssl_ctx`.
+ */
+ outcome::result<void> set(ssl_ctx_st *ssl_ctx);
+
+ private:
+ /**
+ * Validate peer ALPN.
+ */
+ static int select(ssl_st *ssl,
+ const unsigned char **out,
+ unsigned char *outlen,
+ const unsigned char *in,
+ unsigned int inlen,
+ void *void_self);
+
+ qtils::Bytes bytes_;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/config.hpp b/src/snp/connections/config.hpp
new file mode 100644
index 0000000..a8a813e
--- /dev/null
+++ b/src/snp/connections/config.hpp
@@ -0,0 +1,19 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "crypto/ed25519.hpp"
+#include "types/genesis_hash.hpp"
+
+namespace jam::snp {
+ struct ConnectionsConfig {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L30-L35
+ GenesisHash genesis;
+ crypto::ed25519::KeyPair keypair;
+ std::optional<uint16_t> listen_port;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/connection.cpp b/src/snp/connections/connection.cpp
new file mode 100644
index 0000000..c7ad566
--- /dev/null
+++ b/src/snp/connections/connection.cpp
@@ -0,0 +1,39 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/connection.hpp"
+
+#include <boost/asio/dispatch.hpp>
+
+#include "coro/set_thread.hpp"
+#include "snp/connections/error.hpp"
+#include "snp/connections/lsquic/engine.hpp"
+
+namespace jam::snp {
+ using lsquic::Engine;
+
+ Connection::Connection(IoContextPtr io_context_ptr,
+ lsquic::ConnCtx *conn_ctx,
+ ConnectionInfo info)
+ : io_context_ptr_{std::move(io_context_ptr)},
+ conn_ctx_{std::move(conn_ctx)},
+ info_{std::move(info)} {}
+
+ Connection::~Connection() {
+ boost::asio::dispatch(*io_context_ptr_, [conn_ctx{conn_ctx_}] {
+ Engine::destroyConnection(conn_ctx);
+ });
+ }
+
+ const ConnectionInfo &Connection::info() const {
+ return info_;
+ }
+
+ StreamPtrCoroOutcome Connection::open(SelfSPtr self, ProtocolId protocol_id) {
+ co_await setCoroThread(self->io_context_ptr_);
+ co_return co_await Engine::openStream(self->conn_ctx_, protocol_id);
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/connection.hpp b/src/snp/connections/connection.hpp
new file mode 100644
index 0000000..f39f2f6
--- /dev/null
+++ b/src/snp/connections/connection.hpp
@@ -0,0 +1,44 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "coro/coro.hpp"
+#include "coro/io_context_ptr.hpp"
+#include "snp/connections/connection_info.hpp"
+#include "snp/connections/protocol_id.hpp"
+#include "snp/connections/stream_ptr.hpp"
+
+namespace jam::snp::lsquic {
+ struct ConnCtx;
+ class Engine;
+} // namespace jam::snp::lsquic
+
+namespace jam::snp {
+ class Connection {
+ friend lsquic::Engine;
+
+ public:
+ using SelfSPtr = std::shared_ptr<Connection>;
+
+ Connection(IoContextPtr io_context_ptr,
+ lsquic::ConnCtx *conn_ctx,
+ ConnectionInfo info);
+ ~Connection();
+
+ const ConnectionInfo &info() const;
+
+ /**
+ * Open stream with specified `ProtocolId`.
+ */
+ static StreamPtrCoroOutcome open(SelfSPtr self, ProtocolId protocol_id);
+
+ private:
+ IoContextPtr io_context_ptr_;
+ lsquic::ConnCtx *conn_ctx_;
+ ConnectionInfo info_;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/connection_id.hpp b/src/snp/connections/connection_id.hpp
new file mode 100644
index 0000000..040e250
--- /dev/null
+++ b/src/snp/connections/connection_id.hpp
@@ -0,0 +1,17 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstdint>
+
+namespace jam::snp {
+ /**
+ * Is not QUIC connection id.
+ * Used to distinguish connections with same peer key.
+ */
+ using ConnectionId = uint64_t;
+} // namespace jam::snp
diff --git a/src/snp/connections/connection_id_counter.hpp b/src/snp/connections/connection_id_counter.hpp
new file mode 100644
index 0000000..efc416e
--- /dev/null
+++ b/src/snp/connections/connection_id_counter.hpp
@@ -0,0 +1,25 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <atomic>
+#include <memory>
+
+#include "snp/connections/connection_id.hpp"
+
+namespace jam::snp {
+ class ConnectionIdCounter {
+ public:
+ ConnectionId make() {
+ return connection_id_->fetch_add(1);
+ }
+
+ private:
+ using Atomic = std::atomic<ConnectionId>;
+ std::shared_ptr<Atomic> connection_id_ = std::make_shared<Atomic>();
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/connection_info.hpp b/src/snp/connections/connection_info.hpp
new file mode 100644
index 0000000..ac3b02a
--- /dev/null
+++ b/src/snp/connections/connection_info.hpp
@@ -0,0 +1,19 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "snp/connections/connection_id.hpp"
+#include "snp/connections/key.hpp"
+
+namespace jam::snp {
+ struct ConnectionInfo {
+ ConnectionId id;
+ Key key;
+
+ bool operator==(const ConnectionInfo &) const = default;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/connection_ptr.hpp b/src/snp/connections/connection_ptr.hpp
new file mode 100644
index 0000000..12f8d3d
--- /dev/null
+++ b/src/snp/connections/connection_ptr.hpp
@@ -0,0 +1,21 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "coro/coro.hpp"
+
+namespace jam::snp {
+ class Connection;
+} // namespace jam::snp
+
+namespace jam::snp {
+ using ConnectionPtr = std::shared_ptr<Connection>;
+ using ConnectionPtrOutcome = outcome::result<ConnectionPtr>;
+ using ConnectionPtrCoroOutcome = CoroOutcome<ConnectionPtr>;
+} // namespace jam::snp
diff --git a/src/snp/connections/connections.cpp b/src/snp/connections/connections.cpp
new file mode 100644
index 0000000..597d661
--- /dev/null
+++ b/src/snp/connections/connections.cpp
@@ -0,0 +1,154 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/connections.hpp"
+
+#include <TODO_qtils/map_entry.hpp>
+#include <TODO_qtils/variant_get.hpp>
+#include <boost/outcome/try.hpp>
+
+#include "coro/set_thread.hpp"
+#include "coro/spawn.hpp"
+#include "coro/weak.hpp"
+#include "coro/yield.hpp"
+#include "snp/connections/address.hpp"
+#include "snp/connections/connection.hpp"
+#include "snp/connections/controller.hpp"
+#include "snp/connections/error.hpp"
+#include "snp/connections/lsquic/engine.hpp"
+
+namespace jam::snp {
+ inline void todoPreferConnection() {
+ // TODO(turuslan): how to deduplicate connections between two peers?
+ throw std::logic_error{"TODO: prefer connection"};
+ }
+
+ Connections::Connections(IoContextPtr io_context_ptr,
+ std::shared_ptr<log::LoggingSystem> logsys,
+ ConnectionsConfig config)
+ : io_context_ptr_{std::move(io_context_ptr)},
+ logsys_{std::move(logsys)},
+ init_{io_context_ptr_},
+ config_{std::move(config)},
+ key_{crypto::ed25519::get_public(config_.keypair)} {}
+
+ CoroOutcome<void> Connections::init(
+ SelfSPtr self, std::weak_ptr<ConnectionsController> controller) {
+ co_await setCoroThread(self->io_context_ptr_);
+ auto init = self->init_.init();
+ self->controller_ = std::move(controller);
+ BOOST_OUTCOME_CO_TRY(auto certificate, TlsCertificate::make(self->config_));
+ BOOST_OUTCOME_CO_TRY(self->client_,
+ lsquic::Engine::make(self->io_context_ptr_,
+ self->logsys_,
+ self->connection_id_counter_,
+ certificate,
+ std::nullopt,
+ self));
+ if (self->config_.listen_port.has_value()) {
+ BOOST_OUTCOME_CO_TRY(self->server_,
+ lsquic::Engine::make(self->io_context_ptr_,
+ self->logsys_,
+ self->connection_id_counter_,
+ certificate,
+ self->config_.listen_port,
+ self));
+ }
+ init.ready();
+ co_return outcome::success();
+ }
+
+ const Key &Connections::key() const {
+ return key_;
+ }
+
+ ConnectionPtrCoroOutcome Connections::connect(SelfSPtr self,
+ Address address) {
+ co_await setCoroThread(self->io_context_ptr_);
+ if (not co_await self->init_.ready()) {
+ co_return ConnectionsError::CONNECTIONS_INIT;
+ }
+ auto state = qtils::entry(self->connections_, address.key);
+ if (not state) {
+ state.insert(
+ std::make_shared<Connecting::element_type>(self->io_context_ptr_));
+ co_await coroSpawn([self, address, state]() mutable -> Coro<void> {
+ co_await coroYield();
+ auto connection_result = CORO_WEAK_AWAIT(
+ self, self->client_->connect(self->client_, address));
+ auto state = qtils::entry(self->connections_, address.key);
+ if (not state or not std::holds_alternative<Connecting>(*state)) {
+ todoPreferConnection();
+ }
+ auto connecting = std::move(std::get<Connecting>(*state));
+ if (connection_result) {
+ auto &connection = connection_result.value();
+ *state = Connected{connection};
+ if (auto controller = self->controller_.lock()) {
+ controller->onOpen(address.key);
+ }
+ } else {
+ state.erase();
+ }
+ CORO_WEAK_AWAIT_V(
+ self, connecting->set(connecting, std::move(connection_result)));
+ });
+ } else if (auto connected = qtils::variantGet<Connected>(*state)) {
+ co_return *connected;
+ }
+ auto connecting = std::get<Connecting>(*state);
+ self.reset();
+ co_return co_await connecting->get(connecting);
+ }
+
+ Coro<void> Connections::serve(SelfSPtr self,
+ ProtocolId protocol_id,
+ ServeProtocol serve) {
+ co_await setCoroThread(self->io_context_ptr_);
+ qtils::entry(self->protocols_, protocol_id).insert(std::move(serve));
+ }
+
+ void Connections::onConnectionAccept(ConnectionPtr connection) {
+ auto state = entry(connections_, connection->info().key);
+ if (state) {
+ todoPreferConnection();
+ }
+ state.insert(Connected{connection});
+ if (auto controller = controller_.lock()) {
+ controller->onOpen(connection->info().key);
+ }
+ }
+
+ void Connections::onConnectionClose(ConnectionInfo connection_info) {
+ auto state = entry(connections_, connection_info.key);
+ if (not state or not std::holds_alternative<Connected>(*state)
+ or std::get<Connected>(*state)->info() != connection_info) {
+ todoPreferConnection();
+ }
+ state.erase();
+ if (auto controller = controller_.lock()) {
+ controller->onClose(connection_info.key);
+ }
+ }
+
+ void Connections::onStreamAccept(ConnectionPtr connection,
+ ProtocolId protocol_id,
+ StreamPtr stream) {
+ coroSpawn(*io_context_ptr_,
+ [self{shared_from_this()},
+ protocol_id,
+ stream{std::move(stream)},
+ connection_info{connection->info()}]() mutable -> Coro<void> {
+ auto serve_it = qtils::entry(self->protocols_, protocol_id);
+ if (not serve_it) {
+ co_return;
+ }
+ auto serve = *serve_it;
+ std::ignore = CORO_WEAK_AWAIT(
+ self, serve(connection_info, std::move(stream)));
+ });
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/connections.hpp b/src/snp/connections/connections.hpp
new file mode 100644
index 0000000..6d50c32
--- /dev/null
+++ b/src/snp/connections/connections.hpp
@@ -0,0 +1,99 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <unordered_map>
+#include <variant>
+
+#include <qtils/bytes_std_hash.hpp>
+
+#include "coro/coro.hpp"
+#include "coro/init.hpp"
+#include "coro/io_context_ptr.hpp"
+#include "snp/connections/config.hpp"
+#include "snp/connections/connection_id_counter.hpp"
+#include "snp/connections/connection_ptr.hpp"
+#include "snp/connections/key.hpp"
+#include "snp/connections/lsquic/controller.hpp"
+
+namespace jam::log {
+ class LoggingSystem;
+} // namespace jam::log
+
+namespace jam::snp {
+ class Address;
+ class ConnectionsController;
+} // namespace jam::snp
+
+namespace jam::snp::lsquic {
+ class Engine;
+} // namespace jam::snp::lsquic
+
+namespace jam::snp {
+ /**
+ * Initiates and accepts connections with peers.
+ * Prevents duplicate connections with peers.
+ */
+ class Connections : public std::enable_shared_from_this<Connections>,
+ public lsquic::EngineController {
+ public:
+ using SelfSPtr = std::shared_ptr<Connections>;
+
+ Connections(IoContextPtr io_context_ptr,
+ std::shared_ptr<log::LoggingSystem> logsys,
+ ConnectionsConfig config);
+
+ /**
+ * Set controller.
+ * Start quic server and client.
+ */
+ static CoroOutcome<void> init(
+ SelfSPtr self, std::weak_ptr<ConnectionsController> controller);
+
+ const Key &key() const;
+
+ /**
+ * Connect or return existing connection.
+ */
+ static ConnectionPtrCoroOutcome connect(SelfSPtr self, Address address);
+
+ using ServeProtocol =
+ std::function<CoroOutcome<void>(ConnectionInfo, StreamPtr)>;
+ /**
+ * Set callback to handle protocol on server side.
+ */
+ static Coro<void> serve(SelfSPtr self,
+ ProtocolId protocol_id,
+ ServeProtocol serve);
+
+ // EngineController
+ void onConnectionAccept(ConnectionPtr connection) override;
+ void onConnectionClose(ConnectionInfo connection_info) override;
+ void onStreamAccept(ConnectionPtr connection,
+ ProtocolId protocol_id,
+ StreamPtr stream) override;
+
+ private:
+ using Connecting = std::shared_ptr<SharedFuture<ConnectionPtrOutcome>>;
+ using Connected = ConnectionPtr;
+
+ IoContextPtr io_context_ptr_;
+ std::shared_ptr<log::LoggingSystem> logsys_;
+ CoroInit init_;
+ ConnectionsConfig config_;
+ Key key_;
+ std::weak_ptr<ConnectionsController> controller_;
+ std::shared_ptr<lsquic::Engine> client_;
+ std::optional<std::shared_ptr<lsquic::Engine>> server_;
+ std::unordered_map<Key,
+ std::variant<Connecting, Connected>,
+ qtils::BytesStdHash>
+ connections_;
+ std::unordered_map<ProtocolId, ServeProtocol> protocols_;
+ ConnectionIdCounter connection_id_counter_;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/controller.hpp b/src/snp/connections/controller.hpp
new file mode 100644
index 0000000..9cf8962
--- /dev/null
+++ b/src/snp/connections/controller.hpp
@@ -0,0 +1,26 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "snp/connections/key.hpp"
+
+namespace jam::snp {
+ class ConnectionsController {
+ public:
+ virtual ~ConnectionsController() = default;
+
+ /**
+ * There is now some connection with peer.
+ */
+ virtual void onOpen(Key key) {}
+
+ /**
+ * There are no more connections with peer.
+ */
+ virtual void onClose(Key key) {}
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/dns_name.cpp b/src/snp/connections/dns_name.cpp
new file mode 100644
index 0000000..5f82fe8
--- /dev/null
+++ b/src/snp/connections/dns_name.cpp
@@ -0,0 +1,46 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/dns_name.hpp"
+
+#include <cppcodec/detail/base32.hpp>
+#include <cppcodec/detail/codec.hpp>
+
+namespace jam::snp::base32 {
+ constexpr auto kAlphabet = "abcdefghijklmnopqrstuvwxyz234567";
+
+ struct Config {
+ template <typename Codec>
+ using codec_impl = cppcodec::detail::stream_codec<Codec, Config>;
+
+ static constexpr size_t alphabet_size() {
+ return 32;
+ }
+ static constexpr char symbol(cppcodec::detail::alphabet_index_t idx) {
+ return kAlphabet[idx];
+ }
+ static constexpr bool generates_padding() {
+ return false;
+ }
+ };
+
+ inline void encode(std::span<char> out, qtils::BytesIn bytes) {
+ using codec = cppcodec::detail::codec<cppcodec::detail::base32<Config>>;
+ codec::encode(out.data(), out.size(), bytes);
+ }
+} // namespace jam::snp::base32
+
+namespace jam::snp {
+ DnsName::DnsName(const Key &key) {
+ chars[0] = 'e';
+ base32::encode(std::span{chars}.subspan(1), key);
+ }
+
+ outcome::result<void> DnsName::set(x509_st *x509) const {
+ // TODO(turuslan): cert.alt = DnsName(key)
+ return outcome::success();
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/dns_name.hpp b/src/snp/connections/dns_name.hpp
new file mode 100644
index 0000000..a9ee4bc
--- /dev/null
+++ b/src/snp/connections/dns_name.hpp
@@ -0,0 +1,37 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <string_view>
+
+#include <qtils/outcome.hpp>
+
+#include "snp/connections/key.hpp"
+
+struct x509_st;
+
+namespace jam::snp {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L15-L16
+ struct DnsName {
+ explicit DnsName(const Key &key);
+
+ constexpr operator std::string_view() const {
+ return std::string_view{chars.data(), chars.size()};
+ }
+
+ /**
+ * Set `DnsName` as subject alternative name for certificate.
+ */
+ outcome::result<void> set(x509_st *x509) const;
+
+ static constexpr size_t kSize = 53;
+ std::array<char, kSize> chars;
+ };
+ constexpr auto format_as(const DnsName &v) {
+ return v.operator std::string_view();
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/error.hpp b/src/snp/connections/error.hpp
new file mode 100644
index 0000000..99f4761
--- /dev/null
+++ b/src/snp/connections/error.hpp
@@ -0,0 +1,131 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <qtils/enum_error_code.hpp>
+
+namespace jam::snp {
+ enum class OpenSslError : uint8_t {
+ EVP_PKEY_get_raw_public_key,
+ EVP_PKEY_new_raw_private_key,
+ SSL_CTX_set_alpn_protos,
+ SSL_CTX_use_certificate,
+ SSL_CTX_use_PrivateKey,
+ SSL_CTX_set_signing_algorithm_prefs,
+ SSL_CTX_set_verify_algorithm_prefs,
+ SSL_get_peer_certificate,
+ X509_gmtime_adj,
+ X509_set_pubkey,
+ X509_get_pubkey,
+ X509_sign,
+ };
+ Q_ENUM_ERROR_CODE(OpenSslError) {
+ using E = decltype(e);
+ switch (e) {
+ case E::EVP_PKEY_get_raw_public_key:
+ return "EVP_PKEY_get_raw_public_key";
+ case E::EVP_PKEY_new_raw_private_key:
+ return "EVP_PKEY_new_raw_private_key";
+ case E::SSL_CTX_set_alpn_protos:
+ return "SSL_CTX_set_alpn_protos";
+ case E::SSL_CTX_use_certificate:
+ return "SSL_CTX_use_certificate";
+ case E::SSL_CTX_use_PrivateKey:
+ return "SSL_CTX_use_PrivateKey";
+ case E::SSL_CTX_set_signing_algorithm_prefs:
+ return "SSL_CTX_set_signing_algorithm_prefs";
+ case E::SSL_CTX_set_verify_algorithm_prefs:
+ return "SSL_CTX_set_verify_algorithm_prefs";
+ case E::SSL_get_peer_certificate:
+ return "SSL_get_peer_certificate";
+ case E::X509_gmtime_adj:
+ return "X509_gmtime_adj";
+ case E::X509_set_pubkey:
+ return "X509_set_pubkey";
+ case E::X509_get_pubkey:
+ return "X509_get_pubkey";
+ case E::X509_sign:
+ return "X509_sign";
+ }
+ }
+
+ enum class LsQuicError : uint8_t {
+ lsquic_conn_make_stream,
+ lsquic_engine_connect,
+ lsquic_engine_new,
+ lsquic_global_init,
+ };
+ Q_ENUM_ERROR_CODE(LsQuicError) {
+ using E = decltype(e);
+ switch (e) {
+ case E::lsquic_conn_make_stream:
+ return "lsquic_conn_make_stream";
+ case E::lsquic_engine_connect:
+ return "lsquic_engine_connect";
+ case E::lsquic_engine_new:
+ return "lsquic_engine_new";
+ case E::lsquic_global_init:
+ return "lsquic_global_init";
+ }
+ }
+
+ enum class ConnectionsError : uint8_t {
+ CONNECTION_OPEN_CLOSED,
+ CONNECTION_OPEN_DUPLICATE,
+ CONNECTIONS_INIT,
+ ENGINE_CONNECT_ALREADY,
+ ENGINE_CONNECT_CLOSED,
+ ENGINE_CONNECT_KEY_MISMATCH,
+ ENGINE_OPEN_STREAM_ALREADY,
+ ENGINE_OPEN_STREAM_TOO_MANY,
+ HANDSHAKE_FAILED,
+ PROTOCOL_ID_MAKE_INVALID,
+ STREAM_READ_CLOSED,
+ STREAM_READ_DESTROYED,
+ STREAM_READ_PROTOCOL_ID_CLOSED,
+ STREAM_READ_TOO_BIG,
+ STREAM_WRITE_CLOSED,
+ STREAM_WRITE_DESTROYED,
+ };
+ Q_ENUM_ERROR_CODE(ConnectionsError) {
+ using E = decltype(e);
+ switch (e) {
+ case E::CONNECTION_OPEN_CLOSED:
+ return "Connection::open closed";
+ case E::CONNECTION_OPEN_DUPLICATE:
+ return "Connection::open duplicate";
+ case E::CONNECTIONS_INIT:
+ return "Connections::init error";
+ case E::ENGINE_CONNECT_ALREADY:
+ return "Engine::connect already";
+ case E::ENGINE_CONNECT_CLOSED:
+ return "Engine::connect closed";
+ case E::ENGINE_CONNECT_KEY_MISMATCH:
+ return "Engine::connect key mismatch";
+ case E::ENGINE_OPEN_STREAM_ALREADY:
+ return "Engine::openStream already";
+ case E::ENGINE_OPEN_STREAM_TOO_MANY:
+ return "Engine::openStream too many streams";
+ case E::HANDSHAKE_FAILED:
+ return "handshake failed";
+ case E::PROTOCOL_ID_MAKE_INVALID:
+ return "ProtocolId::make invalid";
+ case E::STREAM_READ_CLOSED:
+ return "Stream::read closed";
+ case E::STREAM_READ_DESTROYED:
+ return "Stream::read destroyed";
+ case E::STREAM_READ_PROTOCOL_ID_CLOSED:
+ return "Stream::readProtocolId closed";
+ case E::STREAM_READ_TOO_BIG:
+ return "Stream::read too big";
+ case E::STREAM_WRITE_CLOSED:
+ return "Stream::write closed";
+ case E::STREAM_WRITE_DESTROYED:
+ return "Stream::write destroyed";
+ }
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/key.hpp b/src/snp/connections/key.hpp
new file mode 100644
index 0000000..01217e0
--- /dev/null
+++ b/src/snp/connections/key.hpp
@@ -0,0 +1,14 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "crypto/ed25519.hpp"
+
+namespace jam::snp {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L13-L14
+ using Key = crypto::ed25519::Public;
+} // namespace jam::snp
diff --git a/src/snp/connections/lsquic/controller.hpp b/src/snp/connections/lsquic/controller.hpp
new file mode 100644
index 0000000..10af36d
--- /dev/null
+++ b/src/snp/connections/lsquic/controller.hpp
@@ -0,0 +1,36 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "snp/connections/connection_info.hpp"
+#include "snp/connections/connection_ptr.hpp"
+#include "snp/connections/protocol_id.hpp"
+#include "snp/connections/stream_ptr.hpp"
+
+namespace jam::snp::lsquic {
+ class EngineController {
+ public:
+ virtual ~EngineController() = default;
+
+ /**
+ * Connection was accepted.
+ */
+ virtual void onConnectionAccept(ConnectionPtr connection) {}
+
+ /**
+ * Connection was closed.
+ */
+ virtual void onConnectionClose(ConnectionInfo connection_info) {}
+
+ /**
+ * Stream was accepted.
+ */
+ virtual void onStreamAccept(ConnectionPtr connection,
+ ProtocolId protocol_id,
+ StreamPtr stream) {}
+ };
+} // namespace jam::snp::lsquic
diff --git a/src/snp/connections/lsquic/engine.cpp b/src/snp/connections/lsquic/engine.cpp
new file mode 100644
index 0000000..1e3541c
--- /dev/null
+++ b/src/snp/connections/lsquic/engine.cpp
@@ -0,0 +1,605 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/lsquic/engine.hpp"
+
+#include <boost/outcome/try.hpp>
+#include <qtils/option_take.hpp>
+
+#include "coro/set_thread.hpp"
+#include "coro/spawn.hpp"
+#include "log/logger.hpp"
+#include "snp/connections/config.hpp"
+#include "snp/connections/connection.hpp"
+#include "snp/connections/error.hpp"
+#include "snp/connections/lsquic/controller.hpp"
+#include "snp/connections/lsquic/init.hpp"
+#include "snp/connections/stream.hpp"
+
+// TODO(turuslan): unique streams
+// TODO(turuslan): connection/stream close event lag
+
+namespace jam::snp::lsquic {
+ // TODO(turuslan): config
+ constexpr uint32_t kWindowSize = 64 << 10;
+
+ template <typename T>
+ T::Ls *to_ls(T *ptr) {
+ // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
+ return reinterpret_cast<T::Ls *>(ptr);
+ }
+ template <typename T>
+ T *from_ls(typename T::Ls *ptr) {
+ // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
+ return reinterpret_cast<T *>(ptr);
+ }
+
+ void tryDelete(auto *ptr) {
+ if (not ptr->canDelete()) {
+ return;
+ }
+ // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
+ delete ptr;
+ }
+
+ Socket::endpoint_type make_endpoint(const Address &address) {
+ auto ip = boost::asio::ip::make_address_v6(address.ip);
+ return Socket::endpoint_type{ip, address.port};
+ }
+
+ outcome::result<std::shared_ptr<Engine>> Engine::make(
+ IoContextPtr io_context_ptr,
+ std::shared_ptr<log::LoggingSystem> logsys,
+ ConnectionIdCounter connection_id_counter,
+ TlsCertificate certificate,
+ std::optional<uint16_t> listen_port,
+ std::weak_ptr<EngineController> controller) {
+ OUTCOME_TRY(init());
+
+ uint32_t flags = 0;
+ if (listen_port.has_value()) {
+ flags |= LSENG_SERVER;
+ }
+
+ lsquic_engine_settings settings{};
+ lsquic_engine_init_settings(&settings, flags);
+ settings.es_init_max_stream_data_bidi_remote = kWindowSize;
+ settings.es_init_max_stream_data_bidi_local = kWindowSize;
+
+ static lsquic_stream_if stream_if{};
+ stream_if.on_new_conn = on_new_conn;
+ stream_if.on_conn_closed = on_conn_closed;
+ stream_if.on_hsk_done = on_hsk_done;
+ stream_if.on_new_stream = on_new_stream;
+ stream_if.on_close = on_close;
+ stream_if.on_read = on_read;
+ stream_if.on_write = on_write;
+
+ lsquic_engine_api api{};
+ api.ea_settings = &settings;
+
+ Socket socket{*io_context_ptr};
+ boost::system::error_code ec;
+ socket.open(boost::asio::ip::udp::v6(), ec);
+ if (ec) {
+ return ec;
+ }
+ socket.non_blocking(true, ec);
+ if (ec) {
+ return ec;
+ }
+ if (listen_port.has_value()) {
+ auto ip = boost::asio::ip::address_v6::any();
+ socket.bind({ip, listen_port.value()}, ec);
+ if (ec) {
+ return ec;
+ }
+ }
+ auto socket_local_endpoint = socket.local_endpoint(ec);
+ if (ec) {
+ return ec;
+ }
+ auto self =
+ qtils::MakeSharedPrivate::make<Engine>(io_context_ptr,
+ std::move(logsys),
+ std::move(connection_id_counter),
+ std::move(certificate),
+ std::move(socket),
+ socket_local_endpoint,
+ std::move(controller));
+
+ api.ea_stream_if = &stream_if;
+ api.ea_stream_if_ctx = self.get();
+ api.ea_packets_out = ea_packets_out;
+ api.ea_packets_out_ctx = self.get();
+ api.ea_get_ssl_ctx = ea_get_ssl_ctx;
+
+ self->engine_ = lsquic_engine_new(flags, &api);
+ if (self->engine_ == nullptr) {
+ return LsQuicError::lsquic_engine_new;
+ }
+
+ io_context_ptr->post([weak_self{std::weak_ptr{self}}] {
+ if (auto self = weak_self.lock()) {
+ self->readLoop();
+ }
+ });
+
+ return self;
+ }
+
+ Engine::Engine(qtils::MakeSharedPrivate,
+ IoContextPtr io_context_ptr,
+ std::shared_ptr<log::LoggingSystem> logsys,
+ ConnectionIdCounter connection_id_counter,
+ TlsCertificate &&certificate,
+ Socket &&socket,
+ Socket::endpoint_type socket_local_endpoint,
+ std::weak_ptr<EngineController> controller)
+ : io_context_ptr_{std::move(io_context_ptr)},
+ connection_id_counter_{std::move(connection_id_counter)},
+ certificate_{std::move(certificate)},
+ log_{logsys->getLogger("Engine", "snp")},
+ socket_{std::move(socket)},
+ socket_local_endpoint_{std::move(socket_local_endpoint)},
+ controller_{std::move(controller)},
+ timer_{*io_context_ptr_} {}
+
+ Engine::~Engine() {
+ if (engine_ != nullptr) {
+ boost::asio::dispatch(*io_context_ptr_, [engine{engine_}] {
+ // will call `Engine::on_conn_closed`, `Engine::on_close`.
+ lsquic_engine_destroy(engine);
+ });
+ }
+ }
+
+ ConnectionPtrCoroOutcome Engine::connect(SelfSPtr self, Address address) {
+ co_await setCoroThread(self->io_context_ptr_);
+ if (self->connecting_.has_value()) {
+ co_return ConnectionsError::ENGINE_CONNECT_ALREADY;
+ }
+ co_return co_await coroHandler<ConnectionPtrOutcome>(
+ [&](CoroHandler<ConnectionPtrOutcome> &&handler) {
+ self->connecting_.emplace(Connecting{
+ .address = address,
+ .handler = std::move(handler),
+ });
+ // will call `Engine::ea_get_ssl_ctx`, `Engine::on_new_conn`.
+ lsquic_engine_connect(self->engine_,
+ N_LSQVER,
+ self->socket_local_endpoint_.data(),
+ make_endpoint(address).data(),
+ self.get(),
+ nullptr,
+ nullptr,
+ 0,
+ nullptr,
+ 0,
+ nullptr,
+ 0);
+ if (auto connecting = qtils::optionTake(self->connecting_)) {
+ connecting->handler(LsQuicError::lsquic_engine_connect);
+ }
+ self->wantProcess();
+ });
+ }
+
+ void Engine::wantFlush(StreamCtx *stream_ctx) {
+ if (stream_ctx->want_flush) {
+ return;
+ }
+ stream_ctx->want_flush = true;
+ if (not stream_ctx->stream.has_value()) {
+ return;
+ }
+ want_flush_.emplace_back(stream_ctx->stream.value());
+ wantProcess();
+ }
+
+ void Engine::wantProcess() {
+ if (want_process_) {
+ return;
+ }
+ want_process_ = true;
+ boost::asio::post(*io_context_ptr_, [weak_self{weak_from_this()}] {
+ if (auto self = weak_self.lock()) {
+ self->process();
+ }
+ });
+ }
+
+ void Engine::process() {
+ want_process_ = false;
+ auto want_flush = std::exchange(want_flush_, {});
+ for (auto &weak_stream : want_flush) {
+ auto stream = weak_stream.lock();
+ if (not stream) {
+ continue;
+ }
+ if (not stream->stream_ctx_->ls_stream.has_value()) {
+ continue;
+ }
+ stream->stream_ctx_->want_flush = false;
+ lsquic_stream_flush(stream->stream_ctx_->ls_stream.value());
+ }
+ // will call `Engine::on_new_conn`, `Engine::on_conn_closed`,
+ // `Engine::on_new_stream`, `Engine::on_close`, `Engine::on_read`,
+ // `Engine::on_write`, `Engine::ea_packets_out`.
+ lsquic_engine_process_conns(engine_);
+ int us = 0;
+ if (not lsquic_engine_earliest_adv_tick(engine_, &us)) {
+ return;
+ }
+ timer_.expires_after(std::chrono::microseconds{us});
+ auto cb = [weak_self{weak_from_this()}](boost::system::error_code ec) {
+ auto self = weak_self.lock();
+ if (not self) {
+ return;
+ }
+ if (ec) {
+ return;
+ }
+ self->process();
+ };
+ timer_.async_wait(std::move(cb));
+ }
+
+ void Engine::readLoop() {
+ // https://github.com/cbodley/nexus/blob/d1d8486f713fd089917331239d755932c7c8ed8e/src/socket.cc#L293
+ while (true) {
+ socklen_t len = socket_local_endpoint_.size();
+ auto n = recvfrom(socket_.native_handle(),
+ reading_.buffer.data(),
+ reading_.buffer.size(),
+ 0,
+ reading_.remote_endpoint.data(),
+ &len);
+ if (n == -1) {
+ if (errno == EAGAIN or errno == EWOULDBLOCK) {
+ auto cb =
+ [weak_self{weak_from_this()}](boost::system::error_code ec) {
+ auto self = weak_self.lock();
+ if (not self) {
+ return;
+ }
+ if (ec) {
+ SL_ERROR(self->log_, "udp socket read failed");
+ return;
+ }
+ self->readLoop();
+ };
+ socket_.async_wait(boost::asio::socket_base::wait_read,
+ std::move(cb));
+ }
+ break;
+ }
+ // will call `Engine::on_hsk_done`, `Engine::ea_get_ssl_ctx`.
+ lsquic_engine_packet_in(engine_,
+ reading_.buffer.data(),
+ n,
+ socket_local_endpoint_.data(),
+ reading_.remote_endpoint.data(),
+ this,
+ 0);
+ }
+ process();
+ }
+
+ void Engine::destroyConnection(ConnCtx *conn_ctx) {
+ conn_ctx->connection.reset();
+ if (conn_ctx->ls_conn.has_value()) {
+ lsquic_conn_close(conn_ctx->ls_conn.value());
+ } else {
+ tryDelete(conn_ctx);
+ }
+ }
+
+ StreamPtrCoroOutcome Engine::openStream(ConnCtx *conn_ctx,
+ ProtocolId protocol_id) {
+ if (not conn_ctx->ls_conn) {
+ co_return ConnectionsError::CONNECTION_OPEN_CLOSED;
+ }
+ if (conn_ctx->open_stream) {
+ co_return ConnectionsError::ENGINE_OPEN_STREAM_ALREADY;
+ }
+ if (lsquic_conn_n_avail_streams(conn_ctx->ls_conn.value()) == 0) {
+ co_return ConnectionsError::ENGINE_OPEN_STREAM_TOO_MANY;
+ }
+ conn_ctx->open_stream = nullptr;
+ // will call `Engine::on_new_stream`.
+ lsquic_conn_make_stream(conn_ctx->ls_conn.value());
+ auto stream = qtils::optionTake(conn_ctx->open_stream).value();
+ if (stream == nullptr) {
+ co_return LsQuicError::lsquic_conn_make_stream;
+ }
+ // stream not weak, because no other owners yet
+ BOOST_OUTCOME_CO_TRY(co_await stream->writeProtocolId(protocol_id));
+ co_return stream;
+ }
+
+ void Engine::destroyStream(StreamCtx *stream_ctx) {
+ stream_ctx->stream.reset();
+ if (stream_ctx->ls_stream.has_value()) {
+ lsquic_stream_close(stream_ctx->ls_stream.value());
+ } else {
+ tryDelete(stream_ctx);
+ }
+ }
+
+ void Engine::streamAccept(StreamPtr &&stream) {
+ coroSpawn(*io_context_ptr_,
+ [weak_controller{controller_},
+ stream{std::move(stream)}]() mutable -> CoroOutcome<void> {
+ // stream not weak, because no other owners yet
+ BOOST_OUTCOME_CO_TRY(auto protocol_id,
+ co_await stream->readProtocolId());
+ if (auto controller = weak_controller.lock()) {
+ auto &connection = stream->connection_;
+ controller->onStreamAccept(
+ connection, protocol_id, std::move(stream));
+ }
+ co_return outcome::success();
+ });
+ }
+
+ void Engine::streamShutdownRead(StreamCtx *stream_ctx) {
+ if (stream_ctx->ls_stream.has_value()) {
+ lsquic_stream_shutdown(stream_ctx->ls_stream.value(), SHUT_RD);
+ }
+ }
+
+ void Engine::streamShutdownWrite(StreamCtx *stream_ctx) {
+ if (stream_ctx->ls_stream.has_value()) {
+ lsquic_stream_shutdown(stream_ctx->ls_stream.value(), SHUT_WR);
+ }
+ }
+
+ CoroOutcome<bool> Engine::streamReadRaw(StreamCtx *stream_ctx,
+ qtils::BytesOut message) {
+ if (stream_ctx->reading.has_value()) {
+ throw std::logic_error{"Engine::streamReadRaw duplicate"};
+ }
+ auto remaining = message;
+ while (not remaining.empty()) {
+ [[unlikely]] if (not stream_ctx->ls_stream.has_value()) {
+ co_return ConnectionsError::STREAM_READ_CLOSED;
+ }
+ auto n = lsquic_stream_read(
+ stream_ctx->ls_stream.value(), remaining.data(), remaining.size());
+ if (n == 0) {
+ if (remaining.size() == message.size()) {
+ co_return false;
+ } else {
+ co_return ConnectionsError::STREAM_READ_CLOSED;
+ }
+ }
+ if (n == -1) {
+ if (errno != EWOULDBLOCK) {
+ co_return ConnectionsError::STREAM_READ_CLOSED;
+ }
+ co_await coroHandler<void>([&](CoroHandler<void> &&handler) {
+ stream_ctx->reading.emplace(std::move(handler));
+ lsquic_stream_wantread(stream_ctx->ls_stream.value(), 1);
+ });
+ continue;
+ }
+ remaining = remaining.subspan(n);
+ }
+ co_return true;
+ }
+
+ CoroOutcome<void> Engine::streamWriteRaw(StreamCtx *stream_ctx,
+ qtils::BytesIn message) {
+ if (stream_ctx->writing.has_value()) {
+ throw std::logic_error{"Engine::streamWriteRaw duplicate"};
+ }
+ auto remaining = message;
+ while (not remaining.empty()) {
+ [[unlikely]] if (not stream_ctx->ls_stream.has_value()) {
+ co_return ConnectionsError::STREAM_WRITE_CLOSED;
+ }
+ auto n = lsquic_stream_write(
+ stream_ctx->ls_stream.value(), remaining.data(), remaining.size());
+ if (n < 0) {
+ co_return ConnectionsError::STREAM_WRITE_CLOSED;
+ }
+ if (n != 0) {
+ remaining = remaining.subspan(n);
+ auto self = stream_ctx->engine.lock();
+ if (not self) {
+ co_return ConnectionsError::STREAM_WRITE_CLOSED;
+ }
+ self->wantFlush(stream_ctx);
+ }
+ if (remaining.empty()) {
+ break;
+ }
+ co_await coroHandler<void>([&](CoroHandler<void> &&handler) {
+ stream_ctx->writing.emplace(std::move(handler));
+ lsquic_stream_wantwrite(stream_ctx->ls_stream.value(), 1);
+ });
+ }
+ co_return outcome::success();
+ }
+
+ lsquic_conn_ctx_t *Engine::on_new_conn(void *void_self,
+ lsquic_conn_t *ls_conn) {
+ Engine *self = static_cast<Engine *>(void_self);
+ auto connecting = qtils::optionTake(self->connecting_);
+ auto is_connecting = connecting.has_value();
+ // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
+ auto *conn_ctx = new ConnCtx{
+ .engine = self->weak_from_this(),
+ .ls_conn = ls_conn,
+ .connecting = std::move(connecting),
+ };
+ auto *ls_conn_ctx = to_ls(conn_ctx);
+ lsquic_conn_set_ctx(ls_conn, ls_conn_ctx);
+ if (not is_connecting) {
+ // lsquic doesn't call `on_hsk_done` for incoming connection
+ on_hsk_done(ls_conn, LSQ_HSK_OK);
+ }
+ return ls_conn_ctx;
+ }
+
+ void Engine::on_conn_closed(lsquic_conn_t *ls_conn) {
+ auto *conn_ctx = from_ls<ConnCtx>(lsquic_conn_get_ctx(ls_conn));
+ conn_ctx->ls_conn.reset();
+ lsquic_conn_set_ctx(ls_conn, nullptr);
+ if (auto connecting = qtils::optionTake(conn_ctx->connecting)) {
+ connecting->handler(ConnectionsError::ENGINE_CONNECT_CLOSED);
+ } else if (auto self = conn_ctx->engine.lock()) {
+ if (auto controller = self->controller_.lock()) {
+ controller->onConnectionClose(conn_ctx->info.value());
+ }
+ }
+ tryDelete(conn_ctx);
+ }
+
+ void Engine::on_hsk_done(lsquic_conn_t *ls_conn, lsquic_hsk_status status) {
+ auto *conn_ctx = from_ls<ConnCtx>(lsquic_conn_get_ctx(ls_conn));
+ auto self = conn_ctx->engine.lock();
+ if (not self) {
+ return;
+ }
+ auto ok = status == LSQ_HSK_OK or status == LSQ_HSK_RESUMED_OK;
+ auto connecting = qtils::optionTake(conn_ctx->connecting);
+ auto connection_result = [&]() -> ConnectionPtrOutcome {
+ if (not ok) {
+ return ConnectionsError::HANDSHAKE_FAILED;
+ }
+ OUTCOME_TRY(key, TlsCertificate::get_key(lsquic_conn_ssl(ls_conn)));
+ if (connecting.has_value() and key != connecting->address.key) {
+ return ConnectionsError::ENGINE_CONNECT_KEY_MISMATCH;
+ }
+ conn_ctx->info = ConnectionInfo{
+ .id = self->connection_id_counter_.make(),
+ .key = key,
+ };
+ auto connection = std::make_shared<Connection>(
+ self->io_context_ptr_, conn_ctx, conn_ctx->info.value());
+ conn_ctx->connection = connection;
+ return connection;
+ }();
+ if (not connection_result) {
+ lsquic_conn_close(ls_conn);
+ }
+ if (connecting.has_value()) {
+ connecting->handler(std::move(connection_result));
+ } else if (connection_result) {
+ auto &connection = connection_result.value();
+ if (auto controller = self->controller_.lock()) {
+ controller->onConnectionAccept(std::move(connection));
+ }
+ }
+ }
+
+ lsquic_stream_ctx_t *Engine::on_new_stream(void *void_self,
+ lsquic_stream_t *ls_stream) {
+ Engine *self = static_cast<Engine *>(void_self);
+ auto *conn_ctx =
+ from_ls<ConnCtx>(lsquic_conn_get_ctx(lsquic_stream_conn(ls_stream)));
+ // NOLINTNEXTLINE(cppcoreguidelines-owning-memory)
+ auto *stream_ctx = new StreamCtx{
+ .engine = self->weak_from_this(),
+ .ls_stream = ls_stream,
+ };
+ ConnectionPtr connection;
+ if (conn_ctx->connection.has_value()) {
+ connection = conn_ctx->connection->lock();
+ }
+ if (connection) {
+ auto stream = std::make_shared<Stream>(
+ self->io_context_ptr_, connection, stream_ctx);
+ stream_ctx->stream = stream;
+ if (conn_ctx->open_stream.has_value()) {
+ conn_ctx->open_stream.value() = stream;
+ } else {
+ self->streamAccept(std::move(stream));
+ }
+ } else {
+ lsquic_stream_close(ls_stream);
+ }
+ return to_ls(stream_ctx);
+ }
+
+ void Engine::on_close(lsquic_stream_t *ls_stream,
+ lsquic_stream_ctx_t *ls_stream_ctx) {
+ auto *stream_ctx = from_ls<StreamCtx>(ls_stream_ctx);
+ stream_ctx->ls_stream.reset();
+ if (auto reading = qtils::optionTake(stream_ctx->reading)) {
+ reading.value()();
+ }
+ if (auto writing = qtils::optionTake(stream_ctx->writing)) {
+ writing.value()();
+ }
+ tryDelete(stream_ctx);
+ }
+
+ void Engine::on_read(lsquic_stream_t *ls_stream,
+ lsquic_stream_ctx_t *ls_stream_ctx) {
+ lsquic_stream_wantread(ls_stream, 0);
+ auto *stream_ctx = from_ls<StreamCtx>(ls_stream_ctx);
+ if (auto reading = qtils::optionTake(stream_ctx->reading)) {
+ reading.value()();
+ }
+ }
+
+ void Engine::on_write(lsquic_stream_t *ls_stream,
+ lsquic_stream_ctx_t *ls_stream_ctx) {
+ lsquic_stream_wantwrite(ls_stream, 0);
+ auto *stream_ctx = from_ls<StreamCtx>(ls_stream_ctx);
+ if (auto writing = qtils::optionTake(stream_ctx->writing)) {
+ writing.value()();
+ }
+ }
+
+ ssl_ctx_st *Engine::ea_get_ssl_ctx(void *void_self, const sockaddr *) {
+ Engine *self = static_cast<Engine *>(void_self);
+ return self->certificate_;
+ }
+
+ int Engine::ea_packets_out(void *void_self,
+ const lsquic_out_spec *out_spec,
+ unsigned n_packets_out) {
+ Engine *self = static_cast<Engine *>(void_self);
+ // https://github.com/cbodley/nexus/blob/d1d8486f713fd089917331239d755932c7c8ed8e/src/socket.cc#L218
+ int r = 0;
+ for (auto &spec : std::span{out_spec, n_packets_out}) {
+ msghdr msg{};
+ msg.msg_iov = spec.iov;
+ msg.msg_iovlen = spec.iovlen;
+ // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast)
+ msg.msg_name = const_cast<sockaddr *>(spec.dest_sa);
+ msg.msg_namelen = spec.dest_sa->sa_family == AF_INET
+ ? sizeof(sockaddr_in)
+ : sizeof(sockaddr_in6);
+ auto n = sendmsg(self->socket_.native_handle(), &msg, 0);
+ if (n == -1) {
+ if (errno == EAGAIN or errno == EWOULDBLOCK) {
+ auto cb = [weak_self{self->weak_from_this()}](
+ boost::system::error_code ec) {
+ auto self = weak_self.lock();
+ if (not self) {
+ return;
+ }
+ if (ec) {
+ SL_ERROR(self->log_, "udp socket write failed");
+ return;
+ }
+ // will call `Engine::ea_packets_out`.
+ lsquic_engine_send_unsent_packets(self->engine_);
+ };
+ self->socket_.async_wait(Socket::wait_write, std::move(cb));
+ }
+ break;
+ }
+ ++r;
+ }
+ return r;
+ }
+} // namespace jam::snp::lsquic
diff --git a/src/snp/connections/lsquic/engine.hpp b/src/snp/connections/lsquic/engine.hpp
new file mode 100644
index 0000000..7b75c9e
--- /dev/null
+++ b/src/snp/connections/lsquic/engine.hpp
@@ -0,0 +1,207 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <deque>
+#include <lsquic.h>
+
+#include <TODO_qtils/make_shared_private.hpp>
+#include <boost/asio/ip/udp.hpp>
+#include <boost/asio/steady_timer.hpp>
+
+#include "coro/coro.hpp"
+#include "coro/handler.hpp"
+#include "coro/io_context_ptr.hpp"
+#include "snp/connections/address.hpp"
+#include "snp/connections/connection_id_counter.hpp"
+#include "snp/connections/connection_info.hpp"
+#include "snp/connections/connection_ptr.hpp"
+#include "snp/connections/protocol_id.hpp"
+#include "snp/connections/stream_ptr.hpp"
+#include "snp/connections/tls_certificate.hpp"
+
+struct sockaddr;
+
+namespace soralog {
+ class Logger;
+} // namespace soralog
+
+namespace jam::log {
+ class LoggingSystem;
+} // namespace jam::log
+
+namespace jam::snp {
+ class ConnectionsConfig;
+} // namespace jam::snp
+
+namespace jam::snp::lsquic {
+ class Engine;
+ class EngineController;
+} // namespace jam::snp::lsquic
+
+namespace jam::snp::lsquic {
+ using Socket = boost::asio::ip::udp::socket;
+ Socket::endpoint_type make_endpoint(const Address &address);
+
+ /**
+ * Captures `Engine::connect` arguments.
+ */
+ struct Connecting {
+ Address address;
+ CoroHandler<ConnectionPtrOutcome> handler;
+ };
+
+ /**
+ * `lsquic_conn_ctx_t`.
+ */
+ struct ConnCtx {
+ using Ls = lsquic_conn_ctx_t;
+
+ std::weak_ptr<Engine> engine;
+ std::optional<lsquic_conn_t *> ls_conn;
+ std::optional<std::weak_ptr<Connection>> connection;
+ std::optional<Connecting> connecting;
+ std::optional<ConnectionInfo> info;
+ std::optional<StreamPtr> open_stream;
+
+ bool canDelete() const {
+ return not ls_conn.has_value() and not connection.has_value();
+ }
+ };
+
+ /**
+ * `lsquic_stream_ctx_t`.
+ */
+ struct StreamCtx {
+ using Ls = lsquic_stream_ctx_t;
+
+ std::weak_ptr<Engine> engine;
+ std::optional<lsquic_stream_t *> ls_stream;
+ std::optional<std::weak_ptr<Stream>> stream;
+ std::optional<CoroHandler<void>> reading;
+ std::optional<CoroHandler<void>> writing;
+ bool want_flush = false;
+
+ bool canDelete() const {
+ return not ls_stream.has_value() and not stream.has_value();
+ }
+ };
+
+ class Engine : public std::enable_shared_from_this<Engine> {
+ friend Connection;
+ friend Stream;
+
+ public:
+ using SelfSPtr = std::shared_ptr<Engine>;
+
+ static outcome::result<std::shared_ptr<Engine>> make(
+ IoContextPtr io_context_ptr,
+ std::shared_ptr<log::LoggingSystem> logsys,
+ ConnectionIdCounter connection_id_counter,
+ TlsCertificate certificate,
+ std::optional<uint16_t> listen_port,
+ std::weak_ptr<EngineController> controller);
+ Engine(qtils::MakeSharedPrivate,
+ IoContextPtr io_context_ptr,
+ std::shared_ptr<log::LoggingSystem> logsys,
+ ConnectionIdCounter connection_id_counter,
+ TlsCertificate &&certificate,
+ Socket &&socket,
+ Socket::endpoint_type socket_local_endpoint,
+ std::weak_ptr<EngineController> controller);
+ ~Engine();
+
+ static ConnectionPtrCoroOutcome connect(SelfSPtr self, Address address);
+
+ private:
+ struct Reading {
+ static constexpr size_t kMaxUdpPacketSize = 64 << 10;
+ qtils::BytesN<kMaxUdpPacketSize> buffer;
+ boost::asio::ip::udp::endpoint remote_endpoint;
+ };
+
+ void wantFlush(StreamCtx *stream_ctx);
+ void wantProcess();
+ void process();
+ void readLoop();
+ static void destroyConnection(ConnCtx *conn_ctx);
+ static StreamPtrCoroOutcome openStream(ConnCtx *conn_ctx,
+ ProtocolId protocol_id);
+ static void destroyStream(StreamCtx *stream_ctx);
+ void streamAccept(StreamPtr &&stream);
+ static void streamShutdownRead(StreamCtx *stream_ctx);
+ static void streamShutdownWrite(StreamCtx *stream_ctx);
+ static CoroOutcome<bool> streamReadRaw(StreamCtx *stream_ctx,
+ qtils::BytesOut message);
+ static CoroOutcome<void> streamWriteRaw(StreamCtx *stream_ctx,
+ qtils::BytesIn message);
+
+ /**
+ * Called from `lsquic_engine_connect` (client),
+ * `lsquic_engine_process_conns` (server).
+ */
+ static lsquic_conn_ctx_t *on_new_conn(void *void_self,
+ lsquic_conn_t *ls_conn);
+ /**
+ * Called from `lsquic_engine_process_conns`, `lsquic_engine_destroy`.
+ */
+ static void on_conn_closed(lsquic_conn_t *ls_conn);
+ /**
+ * Called from `lsquic_engine_packet_in` (client),
+ * `on_new_conn` (server).
+ */
+ static void on_hsk_done(lsquic_conn_t *ls_conn, lsquic_hsk_status status);
+ /**
+ * Called from `lsquic_conn_make_stream` (client),
+ * `lsquic_engine_process_conns` (server).
+ */
+ static lsquic_stream_ctx_t *on_new_stream(void *void_self,
+ lsquic_stream_t *ls_stream);
+ /**
+ * Called from `lsquic_engine_process_conns`, `lsquic_engine_destroy`.
+ */
+ static void on_close(lsquic_stream_t *ls_stream,
+ lsquic_stream_ctx_t *ls_stream_ctx);
+ /**
+ * Called from `lsquic_engine_process_conns`.
+ * `lsquic_stream_flush` doesn't work inside `on_read`.
+ */
+ static void on_read(lsquic_stream_t *ls_stream,
+ lsquic_stream_ctx_t *ls_stream_ctx);
+ /**
+ * Called from `lsquic_engine_process_conns`.
+ */
+ static void on_write(lsquic_stream_t *ls_stream,
+ lsquic_stream_ctx_t *ls_stream_ctx);
+ /**
+ * Called from `lsquic_engine_connect` (client),
+ * `lsquic_engine_packet_in` (server).
+ */
+ static ssl_ctx_st *ea_get_ssl_ctx(void *void_self, const sockaddr *);
+ /**
+ * Called from `lsquic_engine_process_conns`,
+ * `lsquic_engine_send_unsent_packets`.
+ */
+ static int ea_packets_out(void *void_self,
+ const lsquic_out_spec *out_spec,
+ unsigned n_packets_out);
+
+ IoContextPtr io_context_ptr_;
+ std::shared_ptr<soralog::Logger> log_;
+ ConnectionIdCounter connection_id_counter_;
+ TlsCertificate certificate_;
+ Socket socket_;
+ Socket::endpoint_type socket_local_endpoint_;
+ std::weak_ptr<EngineController> controller_;
+ boost::asio::steady_timer timer_;
+ lsquic_engine_t *engine_ = nullptr;
+ Reading reading_;
+ std::optional<Connecting> connecting_;
+ std::deque<std::weak_ptr<Stream>> want_flush_;
+ bool want_process_ = false;
+ };
+} // namespace jam::snp::lsquic
diff --git a/src/snp/connections/lsquic/init.cpp b/src/snp/connections/lsquic/init.cpp
new file mode 100644
index 0000000..df6c069
--- /dev/null
+++ b/src/snp/connections/lsquic/init.cpp
@@ -0,0 +1,24 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/lsquic/init.hpp"
+
+#include <lsquic.h>
+
+#include "snp/connections/error.hpp"
+
+namespace jam::snp::lsquic {
+ outcome::result<void> init() {
+ static auto ok = [] {
+ return lsquic_global_init(LSQUIC_GLOBAL_CLIENT | LSQUIC_GLOBAL_SERVER)
+ == 0;
+ }();
+ if (not ok) {
+ return LsQuicError::lsquic_global_init;
+ }
+ return outcome::success();
+ }
+} // namespace jam::snp::lsquic
diff --git a/src/snp/connections/lsquic/init.hpp b/src/snp/connections/lsquic/init.hpp
new file mode 100644
index 0000000..b51dc58
--- /dev/null
+++ b/src/snp/connections/lsquic/init.hpp
@@ -0,0 +1,13 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <qtils/outcome.hpp>
+
+namespace jam::snp::lsquic {
+ outcome::result<void> init();
+} // namespace jam::snp::lsquic
diff --git a/src/snp/connections/lsquic/log.hpp b/src/snp/connections/lsquic/log.hpp
new file mode 100644
index 0000000..a473e22
--- /dev/null
+++ b/src/snp/connections/lsquic/log.hpp
@@ -0,0 +1,47 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <lsquic.h>
+
+#include "log/logger.hpp"
+
+namespace jam::snp::lsquic::log_level {
+ constexpr auto *emerg = "emerg";
+ constexpr auto *alert = "alert";
+ constexpr auto *crit = "crit";
+ constexpr auto *error = "error";
+ constexpr auto *warn = "warn";
+ constexpr auto *notice = "notice";
+ constexpr auto *info = "info";
+ constexpr auto *debug = "debug";
+} // namespace jam::snp::lsquic::log_level
+
+namespace jam::snp::lsquic {
+ /**
+ * Enable lsquic log.
+ */
+ inline void log(std::shared_ptr<soralog::Logger> log,
+ const char *level = log_level::debug) {
+ static std::shared_ptr<soralog::Logger> static_log;
+ static lsquic_logger_if ls_log{
+ +[](void *, const char *buf, size_t len) {
+ if (static_log != nullptr) {
+ std::string_view message{buf, len};
+ while (message.ends_with("\n")) {
+ message.remove_suffix(1);
+ }
+ static_log->info("{}", message);
+ }
+ return 0;
+ },
+ };
+ static_log = std::move(log);
+ lsquic_logger_init(&ls_log, nullptr, LLTS_NONE);
+ lsquic_set_log_level(level);
+ }
+} // namespace jam::snp::lsquic
diff --git a/src/snp/connections/message_size.hpp b/src/snp/connections/message_size.hpp
new file mode 100644
index 0000000..910244e
--- /dev/null
+++ b/src/snp/connections/message_size.hpp
@@ -0,0 +1,17 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <limits>
+
+namespace jam::snp {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L109-L111
+ using MessageSize = uint32_t;
+ constexpr MessageSize kMessageSizeMax =
+ std::numeric_limits<MessageSize>::max();
+} // namespace jam::snp
diff --git a/src/snp/connections/prefer_key.cpp b/src/snp/connections/prefer_key.cpp
new file mode 100644
index 0000000..0326594
--- /dev/null
+++ b/src/snp/connections/prefer_key.cpp
@@ -0,0 +1,13 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/prefer_key.hpp"
+
+namespace jam::snp {
+ bool prefer_key(const Key &a, const Key &b) {
+ return ((a[31] > 127) != (b[31] > 127)) != (a < b);
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/prefer_key.hpp b/src/snp/connections/prefer_key.hpp
new file mode 100644
index 0000000..8dc6c0b
--- /dev/null
+++ b/src/snp/connections/prefer_key.hpp
@@ -0,0 +1,17 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "snp/connections/key.hpp"
+
+namespace jam::snp {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L52-L62
+ /**
+ * Is first key preferred over second.
+ */
+ bool prefer_key(const Key &a, const Key &b);
+} // namespace jam::snp
diff --git a/src/snp/connections/protocol_id.cpp b/src/snp/connections/protocol_id.cpp
new file mode 100644
index 0000000..a234ea6
--- /dev/null
+++ b/src/snp/connections/protocol_id.cpp
@@ -0,0 +1,19 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/protocol_id.hpp"
+
+#include "snp/connections/error.hpp"
+
+namespace jam::snp {
+ outcome::result<ProtocolId> ProtocolId::make(Id id, bool unique) {
+ ProtocolId protocol_id{id};
+ if (unique != protocol_id.unique()) {
+ return ConnectionsError::PROTOCOL_ID_MAKE_INVALID;
+ }
+ return protocol_id;
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/protocol_id.hpp b/src/snp/connections/protocol_id.hpp
new file mode 100644
index 0000000..3530a02
--- /dev/null
+++ b/src/snp/connections/protocol_id.hpp
@@ -0,0 +1,59 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstdint>
+
+#include <TODO_qtils/std_hash_of.hpp>
+#include <qtils/outcome.hpp>
+
+namespace jam::snp {
+ class Stream;
+} // namespace jam::snp
+
+namespace jam::snp {
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L87-L101
+ class ProtocolId {
+ friend Stream;
+
+ using Id = uint8_t;
+
+ ProtocolId(Id id) : id_{id} {}
+
+ public:
+ /**
+ * Construct protocol with specified `id`.
+ * Check expected `unique` consistency with `id` range.
+ */
+ static outcome::result<ProtocolId> make(Id id, bool unique);
+
+ auto &id() const {
+ return id_;
+ }
+
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L87-L101
+ /**
+ * Unique protocols reuse one stream per peer.
+ * Ephemeral protocols may create multiple streams.
+ */
+ bool unique() const {
+ return id() < 128;
+ }
+
+ auto operator<=>(const ProtocolId &) const = default;
+
+ private:
+ Id id_;
+ };
+} // namespace jam::snp
+
+template <>
+struct std::hash<jam::snp::ProtocolId> {
+ size_t operator()(const jam::snp::ProtocolId &v) const {
+ return qtils::stdHashOf(v.id());
+ }
+};
diff --git a/src/snp/connections/stream.cpp b/src/snp/connections/stream.cpp
new file mode 100644
index 0000000..18b9955
--- /dev/null
+++ b/src/snp/connections/stream.cpp
@@ -0,0 +1,111 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/stream.hpp"
+
+#include <boost/asio/dispatch.hpp>
+#include <boost/endian/conversion.hpp>
+#include <boost/outcome/try.hpp>
+
+#include "coro/set_thread.hpp"
+#include "coro/weak.hpp"
+#include "snp/connections/error.hpp"
+#include "snp/connections/lsquic/engine.hpp"
+
+namespace jam::snp {
+ using lsquic::Engine;
+
+ using ProtocolIdBytes = qtils::BytesN<1>;
+ using MessageSizeBytes = qtils::BytesN<sizeof(MessageSize)>;
+
+ Stream::Stream(IoContextPtr io_context_ptr,
+ ConnectionPtr connection,
+ lsquic::StreamCtx *stream_ctx)
+ : io_context_ptr_{std::move(io_context_ptr)},
+ connection_{std::move(connection)},
+ stream_ctx_{std::move(stream_ctx)} {}
+
+ Stream::~Stream() {
+ boost::asio::dispatch(*io_context_ptr_, [stream_ctx{stream_ctx_}] {
+ Engine::destroyStream(stream_ctx);
+ });
+ }
+
+ CoroOutcome<bool> Stream::read(SelfSPtr self,
+ qtils::Bytes &buffer,
+ MessageSize max) {
+ co_await setCoroThread(self->io_context_ptr_);
+ MessageSizeBytes size_bytes;
+ BOOST_OUTCOME_CO_TRY(
+ auto read_size,
+ CORO_WEAK_AWAIT(self,
+ Engine::streamReadRaw(self->stream_ctx_, size_bytes),
+ ConnectionsError::STREAM_READ_DESTROYED));
+ if (not read_size) {
+ co_return false;
+ }
+ auto size = boost::endian::load_little_u32(size_bytes.data());
+ if (size > max) {
+ co_return ConnectionsError::STREAM_READ_TOO_BIG;
+ }
+ buffer.resize(size);
+ BOOST_OUTCOME_CO_TRY(
+ auto read_message,
+ CORO_WEAK_AWAIT(self,
+ Engine::streamReadRaw(self->stream_ctx_, buffer),
+ ConnectionsError::STREAM_READ_DESTROYED));
+ if (not read_message) {
+ co_return ConnectionsError::STREAM_READ_CLOSED;
+ }
+ co_return true;
+ }
+
+ Coro<void> Stream::shutdownRead(SelfSPtr self) {
+ co_await setCoroThread(self->io_context_ptr_);
+ Engine::streamShutdownRead(self->stream_ctx_);
+ co_return;
+ }
+
+ CoroOutcome<void> Stream::write(SelfSPtr self, qtils::BytesIn message) {
+ co_await setCoroThread(self->io_context_ptr_);
+ MessageSizeBytes size_bytes;
+ auto size = message.size();
+ if (size > kMessageSizeMax) {
+ throw std::logic_error{"Stream::write max"};
+ }
+ boost::endian::store_little_u32(size_bytes.data(), size);
+ BOOST_OUTCOME_CO_TRY(
+ CORO_WEAK_AWAIT(self,
+ Engine::streamWriteRaw(self->stream_ctx_, size_bytes),
+ ConnectionsError::STREAM_WRITE_DESTROYED));
+ BOOST_OUTCOME_CO_TRY(
+ CORO_WEAK_AWAIT(self,
+ Engine::streamWriteRaw(self->stream_ctx_, message),
+ ConnectionsError::STREAM_WRITE_DESTROYED));
+ co_return outcome::success();
+ }
+
+ Coro<void> Stream::shutdownWrite(SelfSPtr self) {
+ co_await setCoroThread(self->io_context_ptr_);
+ Engine::streamShutdownWrite(self->stream_ctx_);
+ co_return;
+ }
+
+ CoroOutcome<ProtocolId> Stream::readProtocolId() {
+ ProtocolIdBytes bytes;
+ BOOST_OUTCOME_CO_TRY(auto read,
+ co_await Engine::streamReadRaw(stream_ctx_, bytes));
+ if (not read) {
+ co_return ConnectionsError::STREAM_READ_PROTOCOL_ID_CLOSED;
+ }
+ co_return ProtocolId{bytes[0]};
+ }
+
+ CoroOutcome<void> Stream::writeProtocolId(ProtocolId protocol_id) {
+ ProtocolIdBytes bytes{protocol_id.id()};
+ co_return co_await Engine::streamWriteRaw(stream_ctx_, bytes);
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/stream.hpp b/src/snp/connections/stream.hpp
new file mode 100644
index 0000000..9306ce0
--- /dev/null
+++ b/src/snp/connections/stream.hpp
@@ -0,0 +1,84 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <qtils/bytes.hpp>
+
+#include "coro/coro.hpp"
+#include "coro/io_context_ptr.hpp"
+#include "snp/connections/connection_ptr.hpp"
+#include "snp/connections/message_size.hpp"
+#include "snp/connections/protocol_id.hpp"
+
+namespace jam::snp::lsquic {
+ struct StreamCtx;
+ class Engine;
+} // namespace jam::snp::lsquic
+
+namespace jam::snp {
+ class Stream {
+ friend lsquic::Engine;
+
+ public:
+ using SelfSPtr = std::shared_ptr<Stream>;
+
+ Stream(IoContextPtr io_context_ptr,
+ ConnectionPtr connection,
+ lsquic::StreamCtx *stream_ctx);
+ /**
+ * Will close stream and decrement `Connection` shared use count.
+ */
+ ~Stream();
+
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L109-L111
+ /**
+ * Read whole size prefixed message, no more than `max` bytes.
+ * Returns `true` if message was read, or `false` if fin was received or
+ * stream was closed.
+ */
+ static CoroOutcome<bool> read(SelfSPtr self,
+ qtils::Bytes &buffer,
+ MessageSize max);
+
+ /**
+ * Close reading side of stream.
+ */
+ static Coro<void> shutdownRead(SelfSPtr self);
+
+ // https://github.com/zdave-parity/jam-np/blob/5d374b53578cdd93646e3ee19e2b19ea132317b8/simple.md?plain=1#L109-L111
+ /**
+ * Write while size prefixed message.
+ */
+ static CoroOutcome<void> write(SelfSPtr self, qtils::BytesIn message);
+
+ /**
+ * Write fin.
+ * Closes writing side of stream.
+ */
+ static Coro<void> shutdownWrite(SelfSPtr self);
+
+ private:
+ /**
+ * Read protocol id (server).
+ */
+ CoroOutcome<ProtocolId> readProtocolId();
+ /**
+ * Write protocol id (client).
+ */
+ CoroOutcome<void> writeProtocolId(ProtocolId protocol_id);
+
+ /**
+ * `Stream`, `Engine` operations executed on one `IoContextPtr` thread.
+ */
+ IoContextPtr io_context_ptr_;
+ /**
+ * `Stream` keeps `Connection` shared use count alive.
+ */
+ ConnectionPtr connection_;
+ lsquic::StreamCtx *stream_ctx_;
+ };
+} // namespace jam::snp
diff --git a/src/snp/connections/stream_ptr.hpp b/src/snp/connections/stream_ptr.hpp
new file mode 100644
index 0000000..737f465
--- /dev/null
+++ b/src/snp/connections/stream_ptr.hpp
@@ -0,0 +1,20 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "coro/coro.hpp"
+
+namespace jam::snp {
+ class Stream;
+} // namespace jam::snp
+
+namespace jam::snp {
+ using StreamPtr = std::shared_ptr<Stream>;
+ using StreamPtrCoroOutcome = CoroOutcome<StreamPtr>;
+} // namespace jam::snp
diff --git a/src/snp/connections/tls_certificate.cpp b/src/snp/connections/tls_certificate.cpp
new file mode 100644
index 0000000..1d942f9
--- /dev/null
+++ b/src/snp/connections/tls_certificate.cpp
@@ -0,0 +1,105 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "snp/connections/tls_certificate.hpp"
+
+#include <TODO_qtils/asio_buffer.hpp>
+#include <boost/asio/ssl/context.hpp>
+
+#include "snp/connections/alpn.hpp"
+#include "snp/connections/config.hpp"
+#include "snp/connections/dns_name.hpp"
+#include "snp/connections/error.hpp"
+
+namespace jam::snp {
+ outcome::result<void> set_relative_time(ASN1_TIME *o_time, auto delta) {
+ if (not X509_gmtime_adj(
+ o_time,
+ std::chrono::duration_cast<std::chrono::seconds>(delta).count())) {
+ return OpenSslError::X509_gmtime_adj;
+ }
+ return outcome::success();
+ }
+
+ TlsCertificate::TlsCertificate(const ConnectionsConfig &config)
+ : alpn_{std::make_shared<decltype(alpn_)::element_type>(config.genesis)},
+ context_{std::make_shared<decltype(context_)::element_type>(
+ Context::tlsv13)} {}
+
+ outcome::result<TlsCertificate> TlsCertificate::make(
+ const ConnectionsConfig &config) {
+ TlsCertificate self{config};
+ OUTCOME_TRY(self.alpn_->set(self));
+ self.context_->set_verify_mode(Context::verify_peer
+ | Context::verify_fail_if_no_peer_cert
+ | Context::verify_client_once);
+ self.context_->set_verify_callback(verify);
+ std::array<uint16_t, 1> prefs{SSL_SIGN_ED25519};
+ if (not SSL_CTX_set_signing_algorithm_prefs(
+ self, prefs.data(), prefs.size())) {
+ return OpenSslError::SSL_CTX_set_signing_algorithm_prefs;
+ }
+ if (not SSL_CTX_set_verify_algorithm_prefs(
+ self, prefs.data(), prefs.size())) {
+ return OpenSslError::SSL_CTX_set_verify_algorithm_prefs;
+ }
+ auto secret = crypto::ed25519::get_secret(config.keypair);
+ // `EVP_PKEY_new_raw_private_key` requires seed, but in ed25519 secret=seed
+ bssl::UniquePtr<EVP_PKEY> pkey(EVP_PKEY_new_raw_private_key(
+ EVP_PKEY_ED25519, nullptr, secret.data(), secret.size()));
+ if (not pkey) {
+ return OpenSslError::EVP_PKEY_new_raw_private_key;
+ }
+ if (not SSL_CTX_use_PrivateKey(self, pkey.get())) {
+ return OpenSslError::SSL_CTX_use_PrivateKey;
+ }
+
+ bssl::UniquePtr<X509> x509(X509_new());
+ OUTCOME_TRY(set_relative_time(X509_getm_notBefore(x509.get()),
+ -std::chrono::days{1}));
+ OUTCOME_TRY(set_relative_time(X509_getm_notAfter(x509.get()),
+ std::chrono::years{1}));
+ if (not X509_set_pubkey(x509.get(), pkey.get())) {
+ return OpenSslError::X509_set_pubkey;
+ }
+ OUTCOME_TRY(
+ DnsName{crypto::ed25519::get_public(config.keypair)}.set(x509.get()));
+ if (not X509_sign(x509.get(), pkey.get(), nullptr)) {
+ return OpenSslError::X509_sign;
+ }
+ if (not SSL_CTX_use_certificate(self, x509.get())) {
+ return OpenSslError::SSL_CTX_use_certificate;
+ }
+ return self;
+ }
+
+ TlsCertificate::operator ssl_ctx_st *() const {
+ return context_->native_handle();
+ }
+
+ outcome::result<Key> TlsCertificate::get_key(ssl_st *ssl) {
+ bssl::UniquePtr<X509> x509(SSL_get_peer_certificate(ssl));
+ if (not x509) {
+ return OpenSslError::SSL_get_peer_certificate;
+ }
+ bssl::UniquePtr<EVP_PKEY> pkey(X509_get_pubkey(x509.get()));
+ if (not pkey) {
+ return OpenSslError::X509_get_pubkey;
+ }
+ Key key;
+ size_t key_size = key.size();
+ if (not EVP_PKEY_get_raw_public_key(pkey.get(), key.data(), &key_size)) {
+ return OpenSslError::EVP_PKEY_get_raw_public_key;
+ }
+ return key;
+ }
+
+ bool TlsCertificate::verify(bool, boost::asio::ssl::verify_context &ctx) {
+ X509_STORE_CTX *store_ctx = ctx.native_handle();
+ // TODO(turuslan): DnsName(key) == cert.alt
+ return true;
+ }
+} // namespace jam::snp
diff --git a/src/snp/connections/tls_certificate.hpp b/src/snp/connections/tls_certificate.hpp
new file mode 100644
index 0000000..bd5a2e5
--- /dev/null
+++ b/src/snp/connections/tls_certificate.hpp
@@ -0,0 +1,61 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <memory>
+
+#include <qtils/outcome.hpp>
+
+#include "snp/connections/key.hpp"
+
+struct ssl_st;
+
+namespace boost::asio::ssl {
+ class context;
+ class verify_context;
+} // namespace boost::asio::ssl
+
+namespace jam::snp {
+ struct ConnectionsConfig;
+ class Alpn;
+} // namespace jam::snp
+
+struct ssl_ctx_st;
+
+namespace jam::snp {
+ class TlsCertificate {
+ TlsCertificate(const ConnectionsConfig &config);
+
+ public:
+ /**
+ * Generate self-signed tls certificate.
+ */
+ static outcome::result<TlsCertificate> make(
+ const ConnectionsConfig &config);
+
+ /**
+ * Allows passing `*this` to openssl functions.
+ */
+ operator ssl_ctx_st *() const;
+
+ /**
+ * Get peer key from tls certificate.
+ */
+ static outcome::result<Key> get_key(ssl_st *ssl);
+
+ private:
+ using Context = boost::asio::ssl::context;
+
+ static bool verify(bool, boost::asio::ssl::verify_context &ctx);
+
+ /**
+ * Keeps `Alpn` alive for `SSL_CTX_set_alpn_select_cb`.
+ */
+ std::shared_ptr<Alpn> alpn_;
+ std::shared_ptr<Context> context_;
+ };
+} // namespace jam::snp
diff --git a/src/types/genesis_hash.hpp b/src/types/genesis_hash.hpp
new file mode 100644
index 0000000..8c79800
--- /dev/null
+++ b/src/types/genesis_hash.hpp
@@ -0,0 +1,13 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <qtils/bytes.hpp>
+
+namespace jam {
+ using GenesisHash = qtils::BytesN<32>;
+} // namespace jam
diff --git a/test-vectors/asn1.cmake b/test-vectors/asn1.cmake
index 052640a..f03ed29 100644
--- a/test-vectors/asn1.cmake
+++ b/test-vectors/asn1.cmake
@@ -135,7 +135,6 @@ function(add_test_vector name)
target_link_libraries(${TEST_VECTOR}__transition_test
fmt::fmt
${GTEST_DEPS}
- headers
${TEST_VECTOR}__types
)
add_test(${TEST_VECTOR}__transition_test ${TEST_VECTOR}__transition_test)
diff --git a/vcpkg-overlay/cppcodec.cmake b/vcpkg-overlay/cppcodec.cmake
new file mode 100644
index 0000000..c21ba7c
--- /dev/null
+++ b/vcpkg-overlay/cppcodec.cmake
@@ -0,0 +1,4 @@
+
+find_path(CPPCODEC_INCLUDE_DIRS "cppcodec/base32_crockford.hpp")
+add_library(cppcodec INTERFACE)
+target_include_directories(cppcodec INTERFACE ${CPPCODEC_INCLUDE_DIRS})
diff --git a/vcpkg-overlay/liblsquic/disable-asan.patch b/vcpkg-overlay/liblsquic/disable-asan.patch
new file mode 100644
index 0000000..2b05d0e
--- /dev/null
+++ b/vcpkg-overlay/liblsquic/disable-asan.patch
@@ -0,0 +1,23 @@
+diff --git a/CMakeLists.txt b/CMakeLists.txt
+index 65c4776..5d4086a 100644
+--- a/CMakeLists.txt
++++ b/CMakeLists.txt
+@@ -60,12 +60,12 @@ ENDIF()
+
+ IF(CMAKE_BUILD_TYPE STREQUAL "Debug")
+ SET(MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -O0 -g3")
+- IF(CMAKE_C_COMPILER MATCHES "clang" AND
+- NOT "$ENV{TRAVIS}" MATCHES "^true$" AND
+- NOT "$ENV{EXTRA_CFLAGS}" MATCHES "-fsanitize")
+- SET(MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -fsanitize=address")
+- SET(LIBS ${LIBS} -fsanitize=address)
+- ENDIF()
++ # IF(CMAKE_C_COMPILER MATCHES "clang" AND
++ # NOT "$ENV{TRAVIS}" MATCHES "^true$" AND
++ # NOT "$ENV{EXTRA_CFLAGS}" MATCHES "-fsanitize")
++ # SET(MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -fsanitize=address")
++ # SET(LIBS ${LIBS} -fsanitize=address)
++ # ENDIF()
+ # Uncomment to enable cleartext protocol mode (no crypto):
+ #SET (MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -DLSQUIC_ENABLE_HANDSHAKE_DISABLE=1")
+ ELSE()
diff --git a/vcpkg-overlay/liblsquic/fix-found-boringssl.patch b/vcpkg-overlay/liblsquic/fix-found-boringssl.patch
new file mode 100644
index 0000000..a3a632c
--- /dev/null
+++ b/vcpkg-overlay/liblsquic/fix-found-boringssl.patch
@@ -0,0 +1,53 @@
+diff --git a/CMakeLists.txt b/CMakeLists.txt
+index 5d4086a..e085a83 100644
+--- a/CMakeLists.txt
++++ b/CMakeLists.txt
+@@ -120,10 +120,12 @@ IF(CMAKE_BUILD_TYPE STREQUAL "Debug")
+ SET(MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -Od")
+ #SET (MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -DFIU_ENABLE=1")
+ #SET(LIBS ${LIBS} fiu)
++ SET(LIB_NAME ssld cryptod)
+ ELSE()
+ SET(MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -Ox")
+ # Comment out the following line to compile out debug messages:
+ #SET(MY_CMAKE_FLAGS "${MY_CMAKE_FLAGS} -DLSQUIC_LOWEST_LOG_LEVEL=LSQ_LOG_INFO")
++ SET(LIB_NAME ssl crypto)
+ ENDIF()
+
+ ENDIF() #MSVC
+@@ -191,7 +193,7 @@ IF (NOT DEFINED BORINGSSL_LIB AND DEFINED BORINGSSL_DIR)
+ ELSE()
+
+
+- FOREACH(LIB_NAME ssl crypto)
++ FOREACH(LIB ${LIB_NAME})
+ # If BORINGSSL_LIB is defined, try find each lib. Otherwise, user should define BORINGSSL_LIB_ssl,
+ # BORINGSSL_LIB_crypto and so on explicitly. For example, including boringssl and lsquic both via
+ # add_subdirectory:
+@@ -201,20 +203,20 @@ ELSE()
+ # add_subdirectory(third_party/lsquic)
+ IF (DEFINED BORINGSSL_LIB)
+ IF (CMAKE_SYSTEM_NAME STREQUAL Windows)
+- FIND_LIBRARY(BORINGSSL_LIB_${LIB_NAME}
+- NAMES ${LIB_NAME}
++ FIND_LIBRARY(BORINGSSL_LIB_${LIB}
++ NAMES ${LIB}
+ PATHS ${BORINGSSL_LIB}
+ PATH_SUFFIXES Debug Release MinSizeRel RelWithDebInfo
+ NO_DEFAULT_PATH)
+ ELSE()
+- FIND_LIBRARY(BORINGSSL_LIB_${LIB_NAME}
+- NAMES lib${LIB_NAME}${LIB_SUFFIX}
++ FIND_LIBRARY(BORINGSSL_LIB_${LIB}
++ NAMES lib${LI}${LIB_SUFFIX}
+ PATHS ${BORINGSSL_LIB}
+- PATH_SUFFIXES ${LIB_NAME}
++ PATH_SUFFIXES ${LIB}
+ NO_DEFAULT_PATH)
+ ENDIF()
+ ENDIF()
+- IF(BORINGSSL_LIB_${LIB_NAME})
++ IF(BORINGSSL_LIB_${LIB})
+ MESSAGE(STATUS "Found ${LIB_NAME} library: ${BORINGSSL_LIB_${LIB_NAME}}")
+ ELSE()
+ MESSAGE(FATAL_ERROR "BORINGSSL_LIB_${LIB_NAME} library not found")
diff --git a/vcpkg-overlay/liblsquic/lsquic_conn_ssl.patch b/vcpkg-overlay/liblsquic/lsquic_conn_ssl.patch
new file mode 100644
index 0000000..ae7be54
--- /dev/null
+++ b/vcpkg-overlay/liblsquic/lsquic_conn_ssl.patch
@@ -0,0 +1,80 @@
+diff --git a/include/lsquic.h b/include/lsquic.h
+index 389fbcc..c38d027 100644
+--- a/include/lsquic.h
++++ b/include/lsquic.h
+@@ -1671,6 +1671,10 @@ int lsquic_stream_close(lsquic_stream_t *s);
+ int
+ lsquic_stream_has_unacked_data (lsquic_stream_t *s);
+
++/* Return SSL object associated with this connection */
++struct ssl_st *
++lsquic_conn_ssl(struct lsquic_conn *conn);
++
+ /**
+ * Get certificate chain returned by the server. This can be used for
+ * server certificate verification.
+diff --git a/src/liblsquic/lsquic_conn.c b/src/liblsquic/lsquic_conn.c
+index f76550d..31e5285 100644
+--- a/src/liblsquic/lsquic_conn.c
++++ b/src/liblsquic/lsquic_conn.c
+@@ -128,6 +128,12 @@ lsquic_conn_crypto_alg_keysize (const lsquic_conn_t *lconn)
+ }
+
+
++struct ssl_st *
++lsquic_conn_ssl(struct lsquic_conn *lconn) {
++ return lconn->cn_esf_c->esf_get_ssl(lconn->cn_enc_session);
++}
++
++
+ struct stack_st_X509 *
+ lsquic_conn_get_server_cert_chain (struct lsquic_conn *lconn)
+ {
+diff --git a/src/liblsquic/lsquic_enc_sess.h b/src/liblsquic/lsquic_enc_sess.h
+index f45c15f..3505fbd 100644
+--- a/src/liblsquic/lsquic_enc_sess.h
++++ b/src/liblsquic/lsquic_enc_sess.h
+@@ -115,6 +115,9 @@ struct enc_session_funcs_common
+ (*esf_decrypt_packet)(enc_session_t *, struct lsquic_engine_public *,
+ const struct lsquic_conn *, struct lsquic_packet_in *);
+
++ struct ssl_st *
++ (*esf_get_ssl)(enc_session_t *);
++
+ struct stack_st_X509 *
+ (*esf_get_server_cert_chain) (enc_session_t *);
+
+diff --git a/src/liblsquic/lsquic_enc_sess_ietf.c b/src/liblsquic/lsquic_enc_sess_ietf.c
+index 66329c1..076c4c5 100644
+--- a/src/liblsquic/lsquic_enc_sess_ietf.c
++++ b/src/liblsquic/lsquic_enc_sess_ietf.c
+@@ -2519,6 +2519,13 @@ iquic_esf_global_cleanup (void)
+ }
+
+
++static struct ssl_st *
++iquic_esf_get_ssl(enc_session_t *enc_session_p) {
++ struct enc_sess_iquic *const enc_sess = enc_session_p;
++ return enc_sess->esi_ssl;
++}
++
++
+ static struct stack_st_X509 *
+ iquic_esf_get_server_cert_chain (enc_session_t *enc_session_p)
+ {
+@@ -2744,6 +2751,7 @@ const struct enc_session_funcs_common lsquic_enc_session_common_ietf_v1 =
+ .esf_global_cleanup = iquic_esf_global_cleanup,
+ .esf_global_init = iquic_esf_global_init,
+ .esf_tag_len = IQUIC_TAG_LEN,
++ .esf_get_ssl = iquic_esf_get_ssl,
+ .esf_get_server_cert_chain
+ = iquic_esf_get_server_cert_chain,
+ .esf_get_sni = iquic_esf_get_sni,
+@@ -2763,6 +2771,7 @@ const struct enc_session_funcs_common lsquic_enc_session_common_ietf_v1_no_flush
+ .esf_global_cleanup = iquic_esf_global_cleanup,
+ .esf_global_init = iquic_esf_global_init,
+ .esf_tag_len = IQUIC_TAG_LEN,
++ .esf_get_ssl = iquic_esf_get_ssl,
+ .esf_get_server_cert_chain
+ = iquic_esf_get_server_cert_chain,
+ .esf_get_sni = iquic_esf_get_sni,
diff --git a/vcpkg-overlay/liblsquic/portfile.cmake b/vcpkg-overlay/liblsquic/portfile.cmake
new file mode 100644
index 0000000..3602c59
--- /dev/null
+++ b/vcpkg-overlay/liblsquic/portfile.cmake
@@ -0,0 +1,78 @@
+if(VCPKG_TARGET_IS_WINDOWS)
+ # The lib uses CMAKE_WINDOWS_EXPORT_ALL_SYMBOLS, at least until
+ # https://github.com/litespeedtech/lsquic/pull/371 or similar is merged
+ vcpkg_check_linkage(ONLY_STATIC_LIBRARY)
+endif()
+
+vcpkg_from_github(OUT_SOURCE_PATH SOURCE_PATH
+ REPO litespeedtech/lsquic
+ REF v${VERSION}
+ SHA512 40d742779bfa2dc6fdaf0ee8e9349498d373dcffcc6dd27867c18d87309a288ea6811d693043b5d98364d816b818b49445214497475844201241193c0f37b349
+ HEAD_REF master
+ PATCHES
+ disable-asan.patch
+ fix-found-boringssl.patch
+ lsquic_conn_ssl.patch
+)
+
+# Submodules
+vcpkg_from_github(OUT_SOURCE_PATH LSQPACK_SOURCE_PATH
+ REPO litespeedtech/ls-qpack
+ REF v2.5.3
+ HEAD_REF master
+ SHA512 f90502c763abc84532f33d1b8f952aea7869e4e0c5f6bd344532ddd51c4a180958de4086d88b9ec96673a059c806eec9e70007651d4d4e1a73395919dee47ce0
+)
+if(NOT EXISTS "${SOURCE_PATH}/src/ls-hpack/CMakeLists.txt")
+ file(REMOVE_RECURSE "${SOURCE_PATH}/src/liblsquic/ls-qpack")
+ file(RENAME "${LSQPACK_SOURCE_PATH}" "${SOURCE_PATH}/src/liblsquic/ls-qpack")
+endif()
+
+vcpkg_from_github(OUT_SOURCE_PATH LSHPACK_SOURCE_PATH
+ REPO litespeedtech/ls-hpack
+ REF v2.3.2
+ HEAD_REF master
+ SHA512 45d6c8296e8eee511e6a083f89460d5333fc9a49bc078dac55fdec6c46db199de9f150379f02e054571f954a5e3c79af3864dbc53dc57d10a8d2ed26a92d4278
+)
+if(NOT EXISTS "${SOURCE_PATH}/src/lshpack/CMakeLists.txt")
+ file(REMOVE_RECURSE "${SOURCE_PATH}/src/lshpack")
+ file(RENAME "${LSHPACK_SOURCE_PATH}" "${SOURCE_PATH}/src/lshpack")
+endif()
+
+# Configuration
+vcpkg_find_acquire_program(PERL)
+
+string(COMPARE EQUAL "${VCPKG_LIBRARY_LINKAGE}" "dynamic" LSQUIC_SHARED_LIB)
+
+vcpkg_cmake_configure(
+ SOURCE_PATH "${SOURCE_PATH}"
+ OPTIONS
+ "-DPERL=${PERL}"
+ "-DPERL_EXECUTABLE=${PERL}"
+ "-DLSQUIC_SHARED_LIB=${LSQUIC_SHARED_LIB}"
+ "-DBORINGSSL_INCLUDE=${CURRENT_INSTALLED_DIR}/include"
+ -DLSQUIC_BIN=OFF
+ -DLSQUIC_TESTS=OFF
+ OPTIONS_RELEASE
+ "-DBORINGSSL_LIB=${CURRENT_INSTALLED_DIR}/lib"
+ OPTIONS_DEBUG
+ "-DBORINGSSL_LIB=${CURRENT_INSTALLED_DIR}/debug/lib"
+ -DLSQUIC_DEVEL=ON
+)
+
+vcpkg_cmake_install()
+if(VCPKG_TARGET_IS_WINDOWS)
+ # Upstream removed installation of this header after merging changes
+ file(INSTALL "${SOURCE_PATH}/wincompat/vc_compat.h" DESTINATION "${CURRENT_INSTALLED_DIR}/include/lsquic")
+endif()
+
+vcpkg_cmake_config_fixup(PACKAGE_NAME lsquic)
+
+# Concatenate license files and install
+vcpkg_install_copyright(FILE_LIST
+ "${SOURCE_PATH}/LICENSE"
+ "${SOURCE_PATH}/LICENSE.chrome"
+)
+
+# Remove duplicated include directory
+file(REMOVE_RECURSE "${CURRENT_PACKAGES_DIR}/debug/include")
+
diff --git a/vcpkg-overlay/liblsquic/vcpkg.json b/vcpkg-overlay/liblsquic/vcpkg.json
new file mode 100644
index 0000000..ec90032
--- /dev/null
+++ b/vcpkg-overlay/liblsquic/vcpkg.json
@@ -0,0 +1,25 @@
+{
+ "name": "liblsquic",
+ "version": "3.3.2",
+ "port-version": 1,
+ "description": "An implementation of the QUIC and HTTP/3 protocols.",
+ "homepage": "https://github.com/litespeedtech/lsquic",
+ "license": "MIT AND BSD-3-Clause",
+ "supports": "!x86",
+ "dependencies": [
+ "boringssl",
+ {
+ "name": "getopt",
+ "platform": "windows"
+ },
+ {
+ "name": "vcpkg-cmake",
+ "host": true
+ },
+ {
+ "name": "vcpkg-cmake-config",
+ "host": true
+ },
+ "zlib"
+ ]
+}
diff --git a/vcpkg.json b/vcpkg.json
index c3dbc24..ea9826f 100644
--- a/vcpkg.json
+++ b/vcpkg.json
@@ -2,19 +2,21 @@
"name": "cpp-jam",
"version": "0.0.1",
"dependencies": [
- "qtils",
- "scale",
+ "boost-asio",
+ "boost-beast",
+ "boost-di",
+ "boost-program-options",
+ "cppcodec",
"fmt",
- "soralog",
"kagome-crates",
"libb2",
- "boost-di",
- "boost-program-options",
- "boost-asio",
- "boost-beast",
- "prometheus-cpp"
+ "liblsquic",
+ "prometheus-cpp",
+ "qtils",
+ "scale",
+ "soralog"
],
"features": {
- "test": { "description": "Test", "dependencies": ["gtest"]}
+ "test": { "description": "Test", "dependencies": ["gtest"] }
}
}