diff --git a/packages/streamr-libstreamrproxyclient/CMakeLists.txt b/packages/streamr-libstreamrproxyclient/CMakeLists.txt index 11487c20..9715991e 100644 --- a/packages/streamr-libstreamrproxyclient/CMakeLists.txt +++ b/packages/streamr-libstreamrproxyclient/CMakeLists.txt @@ -173,12 +173,23 @@ if (NOT IOS) add_executable(streamrproxyclient-cpp-wrapper-test ${CMAKE_CURRENT_LIST_DIR}/wrappers/cpp/test/StreamrProxyClientCppWrapperTest.cpp) target_link_directories(streamrproxyclient-cpp-wrapper-test PUBLIC ${CMAKE_CURRENT_LIST_DIR}/build) - + target_include_directories(streamrproxyclient-cpp-wrapper-test PUBLIC ${CMAKE_CURRENT_LIST_DIR}/wrappers/cpp/include) - - target_link_libraries(streamrproxyclient-cpp-wrapper-test + + target_link_libraries(streamrproxyclient-cpp-wrapper-test PRIVATE streamrproxyclient - PRIVATE GTest::gtest + PRIVATE GTest::gtest + PRIVATE GTest::gtest_main + ) + + add_executable(streamrnode-cpp-wrapper-test ${CMAKE_CURRENT_LIST_DIR}/wrappers/cpp/test/StreamrNodeCppWrapperTest.cpp) + target_link_directories(streamrnode-cpp-wrapper-test PUBLIC ${CMAKE_CURRENT_LIST_DIR}/build) + + target_include_directories(streamrnode-cpp-wrapper-test PUBLIC ${CMAKE_CURRENT_LIST_DIR}/wrappers/cpp/include) + + target_link_libraries(streamrnode-cpp-wrapper-test + PRIVATE streamrproxyclient + PRIVATE GTest::gtest PRIVATE GTest::gtest_main ) @@ -189,6 +200,9 @@ if (NOT IOS) # default on the 3-core CI runners. gtest_discover_tests(streamr-streamrnode-test-integration PROPERTIES TIMEOUT 600) gtest_discover_tests(streamrproxyclient-cpp-wrapper-test) + # Networked two-node round-trip, same budget as the streamrnode + # integration tests. + gtest_discover_tests(streamrnode-cpp-wrapper-test PROPERTIES TIMEOUT 600) # DISABLED: these tests build the whole streamr-dev/network TS # monorepo at a 2024 commit (ts-integration/install.sh), which no # longer works on current CI runners. The streamr network TS diff --git a/packages/streamr-libstreamrproxyclient/scripts/run-python-tests.sh b/packages/streamr-libstreamrproxyclient/scripts/run-python-tests.sh new file mode 100755 index 00000000..f4dc8d6a --- /dev/null +++ b/packages/streamr-libstreamrproxyclient/scripts/run-python-tests.sh @@ -0,0 +1,32 @@ +#!/bin/bash +# Stages the freshly built shared library next to the Python package +# sources and runs the wrapper's pytest suite against it. +set -e +cd "$(dirname "$0")/.." +# The unversioned name is a symlink to the CURRENT version (stale +# versioned dylibs may sit next to it after a version bump). +LIB=../../build/packages/streamr-libstreamrproxyclient/libstreamrproxyclient.dylib +if [ ! -e "$LIB" ]; then + LIB=../../build/packages/streamr-libstreamrproxyclient/libstreamrproxyclient.so +fi +if [ ! -e "$LIB" ]; then + LIB="" +fi +if [ -z "$LIB" ]; then + echo "build the monorepo first (libstreamrproxyclient not found)" >&2 + exit 1 +fi +DEST=wrappers/python/src/streamrproxyclient +if [[ "$LIB" == *.dylib ]]; then + cp "$LIB" "$DEST/libstreamrproxyclient.dylib" + trap 'rm -f '"$DEST"'/libstreamrproxyclient.dylib' EXIT +else + cp "$LIB" "$DEST/libstreamrproxyclient.so" + trap 'rm -f '"$DEST"'/libstreamrproxyclient.so' EXIT +fi +if python3 -c "import pytest" 2>/dev/null; then + python3 -m pytest wrappers/python/tests -x -q "$@" +else + echo "pytest not installed; running tests directly" + python3 wrappers/python/tests/test_streamr_node.py +fi diff --git a/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrNode.hpp b/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrNode.hpp new file mode 100644 index 00000000..30bc03a1 --- /dev/null +++ b/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrNode.hpp @@ -0,0 +1,344 @@ +#ifndef STREAMR_NODE_HPP +#define STREAMR_NODE_HPP + +// C++ wrapper of the full-node C API (streamrnode.h) — phase D3b of +// trackerless-network-completion-plan.md. Reuses the shared result/error +// plumbing of StreamrProxyClient.hpp. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "StreamrProxyClient.hpp" + +namespace streamrproxyclient { + +struct ErrorNodeNotFound { + static constexpr const char* name = ERROR_NODE_NOT_FOUND; +}; +struct ErrorNodeNotStarted { + static constexpr const char* name = ERROR_NODE_NOT_STARTED; +}; +struct ErrorNodeAlreadyStarted { + static constexpr const char* name = ERROR_NODE_ALREADY_STARTED; +}; +struct ErrorNodeStopped { + static constexpr const char* name = ERROR_NODE_STOPPED; +}; +struct ErrorInvalidEntryPointUrl { + static constexpr const char* name = ERROR_INVALID_ENTRY_POINT_URL; +}; +struct ErrorSubscriptionNotFound { + static constexpr const char* name = ERROR_SUBSCRIPTION_NOT_FOUND; +}; +struct ErrorNodeOperationFailed { + static constexpr const char* name = ERROR_NODE_OPERATION_FAILED; +}; + +using StreamrNodeErrorCode = std::variant< + ErrorInvalidEthereumAddress, + ErrorInvalidStreamPartId, + ErrorNodeNotFound, + ErrorNodeNotStarted, + ErrorNodeAlreadyStarted, + ErrorNodeStopped, + ErrorInvalidEntryPointUrl, + ErrorSubscriptionNotFound, + ErrorNodeOperationFailed>; + +/** + * @brief Exception thrown by StreamrNode operations. + */ +struct StreamrNodeError : public std::runtime_error { + StreamrNodeErrorCode code; + /** The peer the error concerns; empty when not peer-specific. */ + StreamrProxyAddress peer; + + explicit StreamrNodeError(const StreamrError* error) + : std::runtime_error(error->message), code(getErrorCode(error->code)) { + if (error->peer) { + peer = StreamrProxyAddress::fromCProxy(error->peer); + } + } + +private: + static StreamrNodeErrorCode getErrorCode(const char* code) { + if (strcmp(code, ERROR_INVALID_ETHEREUM_ADDRESS) == 0) { + return ErrorInvalidEthereumAddress{}; + } + if (strcmp(code, ERROR_INVALID_STREAM_PART_ID) == 0) { + return ErrorInvalidStreamPartId{}; + } + if (strcmp(code, ERROR_NODE_NOT_FOUND) == 0) { + return ErrorNodeNotFound{}; + } + if (strcmp(code, ERROR_NODE_NOT_STARTED) == 0) { + return ErrorNodeNotStarted{}; + } + if (strcmp(code, ERROR_NODE_ALREADY_STARTED) == 0) { + return ErrorNodeAlreadyStarted{}; + } + if (strcmp(code, ERROR_NODE_STOPPED) == 0) { + return ErrorNodeStopped{}; + } + if (strcmp(code, ERROR_INVALID_ENTRY_POINT_URL) == 0) { + return ErrorInvalidEntryPointUrl{}; + } + if (strcmp(code, ERROR_SUBSCRIPTION_NOT_FOUND) == 0) { + return ErrorSubscriptionNotFound{}; + } + return ErrorNodeOperationFailed{}; + } +}; + +/** + * @brief Options of a StreamrNode; mirrors StreamrNodeConfig with owned + * storage. Zero-initialized defaults mean: no entry points (the node + * starts a new network and acts as its first entry point), no websocket + * server (client-only operation — the mobile mode). + */ +struct StreamrNodeOptions { + std::vector entryPoints; + uint16_t websocketPort = 0; + std::string websocketHost; /**< empty = default (127.0.0.1) */ + bool acceptProxyConnections = false; +}; + +/** + * @brief RAII wrapper of the full-node C API: create/start/stop a node, + * join stream parts, publish, subscribe with a std::function callback. + * Errors are thrown as StreamrNodeError. + */ +class StreamrNode { +public: + /** + * @brief Message callback of subscribe(). Invoked on an internal + * network thread: return quickly and do NOT call StreamrNode methods + * from inside the callback. Both views are only valid during the + * call. + */ + using MessageCallback = + void(std::string_view streamPartId, std::string_view content); + + explicit StreamrNode( + const std::string& ownEthereumAddress, + const StreamrNodeOptions& options = {}) { + std::vector entryPoints; + entryPoints.reserve(options.entryPoints.size()); + for (const auto& entryPoint : options.entryPoints) { + entryPoints.push_back( + {.websocketUrl = entryPoint.websocketUrl.c_str(), + .ethereumAddress = entryPoint.ethereumAddress.c_str()}); + } + const StreamrNodeConfig config{ + .entryPoints = entryPoints.empty() ? nullptr : entryPoints.data(), + .numEntryPoints = entryPoints.size(), + .websocketPort = options.websocketPort, + .websocketHost = options.websocketHost.empty() + ? nullptr + : options.websocketHost.c_str(), + .acceptProxyConnections = options.acceptProxyConnections}; + const StreamrResult* result = nullptr; + this->nodeHandle = + streamrNodeNew(&result, ownEthereumAddress.c_str(), &config); + checkResult(result); + } + + /** + * @brief Stops the node (if running) and releases it. + */ + ~StreamrNode() { + if (this->nodeHandle != 0) { + const StreamrResult* result = nullptr; + streamrNodeDelete(&result, this->nodeHandle); + streamrResultDelete(result); + } + } + + StreamrNode(const StreamrNode&) = delete; + StreamrNode& operator=(const StreamrNode&) = delete; + StreamrNode(StreamrNode&&) = delete; + StreamrNode& operator=(StreamrNode&&) = delete; + + /** + * @brief Starts the node and joins the layer-0 network; blocks until + * connected. A node cannot be started twice or restarted after stop. + */ + void start() const { + const StreamrResult* result = nullptr; + streamrNodeStart(&result, this->nodeHandle); + checkResult(result); + } + + /** + * @brief Stops the node and leaves the network. Terminal; idempotent. + */ + void stop() const { + const StreamrResult* result = nullptr; + streamrNodeStop(&result, this->nodeHandle); + checkResult(result); + } + + void joinStreamPart( + const std::string& streamPartId, + const std::vector& streamPartEntryPoints = {}) + const { + std::vector entryPoints; + entryPoints.reserve(streamPartEntryPoints.size()); + for (const auto& entryPoint : streamPartEntryPoints) { + entryPoints.push_back( + {.websocketUrl = entryPoint.websocketUrl.c_str(), + .ethereumAddress = entryPoint.ethereumAddress.c_str()}); + } + const StreamrResult* result = nullptr; + streamrNodeJoinStreamPart( + &result, + this->nodeHandle, + streamPartId.c_str(), + entryPoints.empty() ? nullptr : entryPoints.data(), + entryPoints.size()); + checkResult(result); + } + + void leaveStreamPart(const std::string& streamPartId) const { + const StreamrResult* result = nullptr; + streamrNodeLeaveStreamPart( + &result, this->nodeHandle, streamPartId.c_str()); + checkResult(result); + } + + /** + * @brief Publishes a message (joining the stream part first if + * needed). Pass an empty ethereumPrivateKey for an unsigned message; + * otherwise hex without 0x prefix, as in proxyClientPublish. + */ + void publish( + const std::string& streamPartId, + const std::string& content, + const std::string& ethereumPrivateKey = {}) const { + const StreamrResult* result = nullptr; + streamrNodePublish( + &result, + this->nodeHandle, + streamPartId.c_str(), + content.data(), + content.size(), + ethereumPrivateKey.empty() ? nullptr : ethereumPrivateKey.c_str()); + checkResult(result); + } + + /** + * @brief Subscribes to a stream part (joining it first if needed). + * @return A subscription handle for unsubscribe(). + * @details The callback stays owned by this node until the node is + * destroyed (not merely until unsubscribe(): the C API may still + * dispatch a message whose delivery already started when + * unsubscribe returns). + */ + uint64_t subscribe( + const std::string& streamPartId, + std::function callback) { + auto ownedCallback = std::make_unique>( + std::move(callback)); + const StreamrResult* result = nullptr; + const uint64_t subscriptionHandle = streamrNodeSubscribe( + &result, + this->nodeHandle, + streamPartId.c_str(), + &StreamrNode::messageTrampoline, + ownedCallback.get()); + checkResult(result); + this->subscriptionCallbacks.push_back(std::move(ownedCallback)); + return subscriptionHandle; + } + + void unsubscribe(uint64_t subscriptionHandle) const { + const StreamrResult* result = nullptr; + streamrNodeUnsubscribe(&result, this->nodeHandle, subscriptionHandle); + checkResult(result); + } + + [[nodiscard]] uint64_t neighborCount( + const std::string& streamPartId) const { + const StreamrResult* result = nullptr; + const uint64_t count = streamrNodeGetNeighborCount( + &result, this->nodeHandle, streamPartId.c_str()); + checkResult(result); + return count; + } + + /** + * @brief Connects the stream part through proxies instead of joining + * its topology; an empty proxy list disconnects it. connectionCount 0 + * means "connect to all given proxies". + */ + void setProxies( + const std::string& streamPartId, + const std::vector& proxies, + StreamrProxyDirection direction, + uint64_t connectionCount = 0) const { + std::vector cProxies; + cProxies.reserve(proxies.size()); + for (const auto& proxy : proxies) { + cProxies.push_back( + {.websocketUrl = proxy.websocketUrl.c_str(), + .ethereumAddress = proxy.ethereumAddress.c_str()}); + } + const StreamrResult* result = nullptr; + streamrNodeSetProxies( + &result, + this->nodeHandle, + streamPartId.c_str(), + cProxies.empty() ? nullptr : cProxies.data(), + cProxies.size(), + direction, + connectionCount); + checkResult(result); + } + +private: + static SharedLibraryWrapper sharedLibraryWrapper; + uint64_t nodeHandle = 0; + std::vector>> + subscriptionCallbacks; + + static void messageTrampoline( + uint64_t /*nodeHandle*/, + const char* streamPartId, + const char* content, + uint64_t contentLength, + void* userData) { + const auto* callback = + static_cast*>(userData); + (*callback)( + std::string_view(streamPartId), + std::string_view(content, contentLength)); + } + + // Throws the first error (deleting the result first); frees the + // result in every case. + static void checkResult(const StreamrResult* result) { + if (result != nullptr && result->numErrors > 0) { + const auto error = StreamrNodeError(&result->errors[0]); + streamrResultDelete(result); + // The result must be freed BEFORE throwing, so the error is + // built first and thrown as a copy — the same pattern the + // proxy-client wrapper uses. + // NOLINTNEXTLINE(misc-throw-by-value-catch-by-reference) + throw error; + } + streamrResultDelete(result); + } +}; + +// The single definition of the static member for header-only use. +inline SharedLibraryWrapper StreamrNode::sharedLibraryWrapper; // NOLINT + +} // namespace streamrproxyclient +#endif // STREAMR_NODE_HPP diff --git a/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp b/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp index d0c0e816..0814871d 100644 --- a/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp +++ b/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp @@ -206,7 +206,7 @@ class StreamrProxyClient { [[nodiscard]] StreamrProxyResult connect( const std::vector& proxies) const { const StreamrResult* result; - std::vector cProxies; + std::vector cProxies; cProxies.reserve(proxies.size()); for (const auto& proxy : proxies) { cProxies.push_back( diff --git a/packages/streamr-libstreamrproxyclient/wrappers/cpp/test/StreamrNodeCppWrapperTest.cpp b/packages/streamr-libstreamrproxyclient/wrappers/cpp/test/StreamrNodeCppWrapperTest.cpp new file mode 100644 index 00000000..0777051f --- /dev/null +++ b/packages/streamr-libstreamrproxyclient/wrappers/cpp/test/StreamrNodeCppWrapperTest.cpp @@ -0,0 +1,134 @@ +// Mirrors StreamrNodeTest.TwoNodesExchangeMessages through the C++ +// wrapper (phase D3b): create/start/join/publish/subscribe round-trip +// between two wrapper-created nodes, plus the error mapping. + +#include +#include +#include +#include +#include +#include +#include + +#include "StreamrNode.hpp" + +using streamrproxyclient::ErrorInvalidEthereumAddress; +using streamrproxyclient::ErrorSubscriptionNotFound; +using streamrproxyclient::StreamrNode; +using streamrproxyclient::StreamrNodeError; +using streamrproxyclient::StreamrNodeOptions; +using streamrproxyclient::StreamrProxyAddress; + +namespace { + +constexpr auto pollInterval = std::chrono::milliseconds(200); +constexpr auto topologyTimeout = std::chrono::seconds(60); +constexpr auto messageTimeout = std::chrono::seconds(30); +constexpr uint16_t entryPointPort = 44461; +constexpr const char* ethereumAddressA = + "0x1234567890123456789012345678901234567890"; +constexpr const char* ethereumAddressB = + "0x1234567890123456789012345678901234567892"; +constexpr const char* ethereumPrivateKey = + "1111111111111111111111111111111111111111111111111111111111111111"; +constexpr const char* streamPartId = + "0xa000000000000000000000000000000000000000#01"; + +template +bool waitUntil(const ConditionType& condition, std::chrono::seconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + if (condition()) { + return true; + } + std::this_thread::sleep_for(pollInterval); + } + return condition(); +} + +struct ReceivedMessages { + std::mutex mutex; + std::vector messages; + + [[nodiscard]] bool contains(const std::string& message) { + std::scoped_lock lock(this->mutex); + return std::ranges::find(this->messages, message) != + this->messages.end(); + } +}; + +} // namespace + +TEST(StreamrNodeCppWrapperTest, InvalidEthereumAddressThrows) { + try { + StreamrNode node("not-an-address"); + FAIL() << "expected StreamrNodeError"; + } catch (const StreamrNodeError& error) { + EXPECT_TRUE( + std::holds_alternative(error.code)); + } +} + +TEST(StreamrNodeCppWrapperTest, TwoNodesExchangeMessages) { + // Node A runs the websocket server of a new network; node B joins + // through it. + StreamrNode nodeA( + ethereumAddressA, StreamrNodeOptions{.websocketPort = entryPointPort}); + nodeA.start(); + + StreamrNode nodeB( + ethereumAddressB, + StreamrNodeOptions{ + .entryPoints = {StreamrProxyAddress{ + .websocketUrl = + "ws://127.0.0.1:" + std::to_string(entryPointPort), + .ethereumAddress = ethereumAddressA}}}); + nodeB.start(); + + ReceivedMessages receivedA; + ReceivedMessages receivedB; + const auto subscriptionA = nodeA.subscribe( + streamPartId, + [&receivedA]( + std::string_view /*streamPartId*/, std::string_view content) { + std::scoped_lock lock(receivedA.mutex); + receivedA.messages.emplace_back(content); + }); + const auto subscriptionB = nodeB.subscribe( + streamPartId, + [&receivedB]( + std::string_view /*streamPartId*/, std::string_view content) { + std::scoped_lock lock(receivedB.mutex); + receivedB.messages.emplace_back(content); + }); + + ASSERT_TRUE(waitUntil( + [&]() { + return nodeA.neighborCount(streamPartId) >= 1 && + nodeB.neighborCount(streamPartId) >= 1; + }, + topologyTimeout)); + + const std::string messageFromA = "hello from A (signed)"; + nodeA.publish(streamPartId, messageFromA, ethereumPrivateKey); + EXPECT_TRUE(waitUntil( + [&]() { return receivedB.contains(messageFromA); }, messageTimeout)); + + const std::string messageFromB = "hello from B (unsigned)"; + nodeB.publish(streamPartId, messageFromB); + EXPECT_TRUE(waitUntil( + [&]() { return receivedA.contains(messageFromB); }, messageTimeout)); + + nodeB.unsubscribe(subscriptionB); + try { + nodeB.unsubscribe(subscriptionB); + FAIL() << "expected StreamrNodeError"; + } catch (const StreamrNodeError& error) { + EXPECT_TRUE( + std::holds_alternative(error.code)); + } + nodeA.unsubscribe(subscriptionA); + + nodeB.stop(); + nodeA.stop(); +} diff --git a/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py b/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py index 03213b69..a4435862 100644 --- a/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py +++ b/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py @@ -272,3 +272,231 @@ def publish(self, data: bytes, ethereumPrivateKey: str = None) -> ProxyClientRes res = ProxyClientResult(result) self.lib.streamrResultDelete(result) return res + + +# --------------------------------------------------------------------------- +# Full-node API (streamrnode.h) — phase D3b. +# --------------------------------------------------------------------------- + +class NodeErrorCodes: + """ + Error codes of the full-node API (streamrnode.h). + """ + ERROR_NODE_NOT_FOUND = "NODE_NOT_FOUND" + ERROR_NODE_NOT_STARTED = "NODE_NOT_STARTED" + ERROR_NODE_ALREADY_STARTED = "NODE_ALREADY_STARTED" + ERROR_NODE_STOPPED = "NODE_STOPPED" + ERROR_INVALID_ENTRY_POINT_URL = "INVALID_ENTRY_POINT_URL" + ERROR_SUBSCRIPTION_NOT_FOUND = "SUBSCRIPTION_NOT_FOUND" + ERROR_NODE_OPERATION_FAILED = "NODE_OPERATION_FAILED" + + +class CStreamrNodeConfig(ctypes.Structure): + """ + C struct for StreamrNodeConfig. + """ + _fields_ = [("entryPoints", ctypes.POINTER(CProxy)), + ("numEntryPoints", ctypes.c_uint64), + ("websocketPort", ctypes.c_uint16), + ("websocketHost", ctypes.c_char_p), + ("acceptProxyConnections", ctypes.c_bool)] + + +# void (*StreamrNodeMessageCallback)(uint64_t nodeHandle, +# const char* streamPartId, const char* content, +# uint64_t contentLength, void* userData) +# content is opaque bytes (may contain NULs), so it is bound as a raw +# pointer and sliced with contentLength. +CMessageCallback = ctypes.CFUNCTYPE( + None, ctypes.c_uint64, ctypes.c_char_p, ctypes.POINTER(ctypes.c_char), + ctypes.c_uint64, ctypes.c_void_p) + + +def _register_node_functions(lib): + """ + Define the ctypes signatures of the streamrNode* functions. + """ + result_out = ctypes.POINTER(ctypes.POINTER(CProxyClientResult)) + lib.streamrNodeNew.argtypes = [ + result_out, ctypes.c_char_p, ctypes.POINTER(CStreamrNodeConfig)] + lib.streamrNodeNew.restype = ctypes.c_uint64 + lib.streamrNodeDelete.argtypes = [result_out, ctypes.c_uint64] + lib.streamrNodeDelete.restype = None + lib.streamrNodeStart.argtypes = [result_out, ctypes.c_uint64] + lib.streamrNodeStart.restype = None + lib.streamrNodeStop.argtypes = [result_out, ctypes.c_uint64] + lib.streamrNodeStop.restype = None + lib.streamrNodeJoinStreamPart.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_char_p, + ctypes.POINTER(CProxy), ctypes.c_uint64] + lib.streamrNodeJoinStreamPart.restype = None + lib.streamrNodeLeaveStreamPart.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_char_p] + lib.streamrNodeLeaveStreamPart.restype = None + lib.streamrNodePublish.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_char_p, ctypes.c_char_p, + ctypes.c_uint64, ctypes.c_char_p] + lib.streamrNodePublish.restype = None + lib.streamrNodeSubscribe.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_char_p, CMessageCallback, + ctypes.c_void_p] + lib.streamrNodeSubscribe.restype = ctypes.c_uint64 + lib.streamrNodeUnsubscribe.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_uint64] + lib.streamrNodeUnsubscribe.restype = None + lib.streamrNodeGetNeighborCount.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_char_p] + lib.streamrNodeGetNeighborCount.restype = ctypes.c_uint64 + lib.streamrNodeSetProxies.argtypes = [ + result_out, ctypes.c_uint64, ctypes.c_char_p, + ctypes.POINTER(CProxy), ctypes.c_uint64, ctypes.c_int, + ctypes.c_uint64] + lib.streamrNodeSetProxies.restype = None + + +class StreamrNode: + """ + The full network node (context manager). Mirrors the C++ + StreamrNode wrapper: create with an optional entry-point list and + websocket port, then start/join/publish/subscribe. + """ + + PROXY_DIRECTION_PUBLISH = 0 + PROXY_DIRECTION_SUBSCRIBE = 1 + + def __init__(self, lib: LibStreamrProxyClient, own_ethereum_address: str, + entry_points: list[Proxy] = None, websocket_port: int = 0, + websocket_host: str = None, + accept_proxy_connections: bool = False): + self.lib = lib.lib + _register_node_functions(self.lib) + self.own_ethereum_address = own_ethereum_address + self.entry_points = entry_points or [] + self.websocket_port = websocket_port + self.websocket_host = websocket_host + self.accept_proxy_connections = accept_proxy_connections + self.node_handle = 0 + # ctypes callback objects must outlive their subscriptions; the C + # API may still dispatch a message whose delivery already started + # when unsubscribe returns, so these are kept until node deletion. + self._callbacks = [] + + def _check(self, result): + if result and result.contents.numErrors > 0: + error = Error(result.contents.errors[0]) + self.lib.streamrResultDelete(result) + raise ProxyClientException(error) + self.lib.streamrResultDelete(result) + + def __enter__(self): + entry_point_array = (CProxy * len(self.entry_points))( + *[CProxy(ep.websocket_url.encode('utf-8'), + ep.ethereum_address.encode('utf-8')) + for ep in self.entry_points]) + config = CStreamrNodeConfig( + entryPoints=entry_point_array if self.entry_points else None, + numEntryPoints=len(self.entry_points), + websocketPort=self.websocket_port, + websocketHost=self.websocket_host.encode('utf-8') + if self.websocket_host else None, + acceptProxyConnections=self.accept_proxy_connections) + result = ctypes.POINTER(CProxyClientResult)() + self.node_handle = self.lib.streamrNodeNew( + ctypes.byref(result), + self.own_ethereum_address.encode('utf-8'), + ctypes.byref(config)) + self._check(result) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeDelete(ctypes.byref(result), self.node_handle) + self.lib.streamrResultDelete(result) + self._callbacks.clear() + + def start(self): + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeStart(ctypes.byref(result), self.node_handle) + self._check(result) + + def stop(self): + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeStop(ctypes.byref(result), self.node_handle) + self._check(result) + + def join_stream_part(self, stream_part_id: str, + entry_points: list[Proxy] = None): + entry_points = entry_points or [] + entry_point_array = (CProxy * len(entry_points))( + *[CProxy(ep.websocket_url.encode('utf-8'), + ep.ethereum_address.encode('utf-8')) + for ep in entry_points]) + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeJoinStreamPart( + ctypes.byref(result), self.node_handle, + stream_part_id.encode('utf-8'), + entry_point_array if entry_points else None, len(entry_points)) + self._check(result) + + def leave_stream_part(self, stream_part_id: str): + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeLeaveStreamPart( + ctypes.byref(result), self.node_handle, + stream_part_id.encode('utf-8')) + self._check(result) + + def publish(self, stream_part_id: str, data: bytes, + ethereum_private_key: str = None): + result = ctypes.POINTER(CProxyClientResult)() + key = ethereum_private_key.encode('utf-8') \ + if ethereum_private_key else None + self.lib.streamrNodePublish( + ctypes.byref(result), self.node_handle, + stream_part_id.encode('utf-8'), data, len(data), key) + self._check(result) + + def subscribe(self, stream_part_id: str, callback) -> int: + """ + Subscribe to a stream part. callback(stream_part_id: str, + content: bytes) is invoked on an internal network thread: return + quickly and do not call node methods from inside it. + """ + def trampoline(_node_handle, c_stream_part_id, c_content, + content_length, _user_data): + callback(c_stream_part_id.decode('utf-8'), + ctypes.string_at(c_content, content_length)) + c_callback = CMessageCallback(trampoline) + result = ctypes.POINTER(CProxyClientResult)() + subscription_handle = self.lib.streamrNodeSubscribe( + ctypes.byref(result), self.node_handle, + stream_part_id.encode('utf-8'), c_callback, None) + self._check(result) + self._callbacks.append(c_callback) + return subscription_handle + + def unsubscribe(self, subscription_handle: int): + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeUnsubscribe( + ctypes.byref(result), self.node_handle, subscription_handle) + self._check(result) + + def neighbor_count(self, stream_part_id: str) -> int: + result = ctypes.POINTER(CProxyClientResult)() + count = self.lib.streamrNodeGetNeighborCount( + ctypes.byref(result), self.node_handle, + stream_part_id.encode('utf-8')) + self._check(result) + return count + + def set_proxies(self, stream_part_id: str, proxies: list[Proxy], + direction: int, connection_count: int = 0): + proxy_array = (CProxy * len(proxies))( + *[CProxy(p.websocket_url.encode('utf-8'), + p.ethereum_address.encode('utf-8')) for p in proxies]) + result = ctypes.POINTER(CProxyClientResult)() + self.lib.streamrNodeSetProxies( + ctypes.byref(result), self.node_handle, + stream_part_id.encode('utf-8'), + proxy_array if proxies else None, len(proxies), direction, + connection_count) + self._check(result) diff --git a/packages/streamr-libstreamrproxyclient/wrappers/python/tests/test_streamr_node.py b/packages/streamr-libstreamrproxyclient/wrappers/python/tests/test_streamr_node.py new file mode 100644 index 00000000..b8dbba84 --- /dev/null +++ b/packages/streamr-libstreamrproxyclient/wrappers/python/tests/test_streamr_node.py @@ -0,0 +1,102 @@ +# Round-trip test of the Python StreamrNode wrapper (phase D3b): mirrors +# StreamrNodeTest.TwoNodesExchangeMessages. Needs the built shared +# library next to the package sources (hatch_build.py copies it for +# package builds; for a repo checkout run scripts/run-python-tests.sh, +# which stages the dylib and invokes pytest). +import sys +import time +import threading +import os + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + +from streamrproxyclient.libstreamrproxyclient import ( # noqa: E402 + LibStreamrProxyClient, + Proxy, + ProxyClientException, + StreamrNode, +) + +ETHEREUM_ADDRESS_A = "0x1234567890123456789012345678901234567890" +ETHEREUM_ADDRESS_B = "0x1234567890123456789012345678901234567892" +ETHEREUM_PRIVATE_KEY = ( + "1111111111111111111111111111111111111111111111111111111111111111") +STREAM_PART_ID = "0xa000000000000000000000000000000000000000#01" +ENTRY_POINT_PORT = 44471 +TOPOLOGY_TIMEOUT_S = 60 +MESSAGE_TIMEOUT_S = 30 + + +def wait_until(condition, timeout_s): + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(0.2) + return condition() + + +def test_invalid_ethereum_address_raises(): + with LibStreamrProxyClient() as lib: + try: + with StreamrNode(lib, "not-an-address"): + raise AssertionError("expected ProxyClientException") + except ProxyClientException as e: + assert e.error.code == "INVALID_ETHEREUM_ADDRESS" + + +def test_two_nodes_exchange_messages(): + with LibStreamrProxyClient() as lib: + with StreamrNode(lib, ETHEREUM_ADDRESS_A, + websocket_port=ENTRY_POINT_PORT) as node_a: + node_a.start() + entry_point = Proxy( + f"ws://127.0.0.1:{ENTRY_POINT_PORT}", ETHEREUM_ADDRESS_A) + with StreamrNode(lib, ETHEREUM_ADDRESS_B, + entry_points=[entry_point]) as node_b: + node_b.start() + + received_a = [] + received_b = [] + lock = threading.Lock() + + def on_message_a(_stream_part_id, content): + with lock: + received_a.append(content) + + def on_message_b(_stream_part_id, content): + with lock: + received_b.append(content) + + sub_a = node_a.subscribe(STREAM_PART_ID, on_message_a) + sub_b = node_b.subscribe(STREAM_PART_ID, on_message_b) + + assert wait_until( + lambda: node_a.neighbor_count(STREAM_PART_ID) >= 1 + and node_b.neighbor_count(STREAM_PART_ID) >= 1, + TOPOLOGY_TIMEOUT_S) + + node_a.publish(STREAM_PART_ID, b"hello from A", + ETHEREUM_PRIVATE_KEY) + assert wait_until( + lambda: b"hello from A" in received_b, + MESSAGE_TIMEOUT_S) + + node_b.publish(STREAM_PART_ID, b"hello from B") + assert wait_until( + lambda: b"hello from B" in received_a, + MESSAGE_TIMEOUT_S) + + node_b.unsubscribe(sub_b) + node_a.unsubscribe(sub_a) + node_b.stop() + node_a.stop() + + +if __name__ == "__main__": + # Direct run without pytest (scripts/run-python-tests.sh falls back + # to this when pytest is unavailable). + test_invalid_ethereum_address_raises() + print("test_invalid_ethereum_address_raises PASSED") + test_two_nodes_exchange_messages() + print("test_two_nodes_exchange_messages PASSED")