From 6a5a415d9df17fec256373a4bb0e0066527cc6ee Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Fri, 17 Jul 2026 11:03:12 +0300 Subject: [PATCH 1/5] Phase D3a: rename the shared C API infrastructure to streamr* names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The proxy-prefixed shared infrastructure both APIs use moves to neutral names in a new streamrcommon.h, per phase D3 of the completion plan (the naming decision deferred from the D1 review): - struct Proxy -> struct StreamrPeer (StreamrEntryPoint stays as an alias of it in streamrnode.h) - struct Error -> struct StreamrError; its 'proxy' member is now 'peer', with the old spelling kept usable through an anonymous union (same storage) - struct ProxyResult -> struct StreamrResult - proxyClientResultDelete -> streamrResultDelete - proxyClientInitLibrary -> streamrInitLibrary - proxyClientCleanupLibrary -> streamrCleanupLibrary The old type names remain as typedef aliases and the old C symbols stay exported (delegating to the new ones), marked deprecated for one release, so existing proxy-API consumers keep compiling and linking. The shared-library major version bumps to 3.0.0 (SOVERSION 3). The proxy client calls (proxyClientNew/Delete/Connect/Publish) keep their names — the API is about proxies; only the shared types in their signatures changed spelling. In-repo consumers (C++ wrapper header, Python wrapper bindings, the integration/end-to-end/ts-interop tests) now use the final names, so the D3b wrapper extensions are written against them once. A dedicated test exercises the deprecated aliases and the pre-3.0 member spelling. All 11 proxy-client integration tests and 9 streamrnode integration tests pass in the per-test-process mode CI uses. (Running several tests in ONE process crashes in the second test — a pre-existing library re-init-after-cleanup bug, unaffected by the rename; follow-up filed.) Co-Authored-By: Claude Fable 5 --- .../VERSION.cmake | 4 +- .../include/streamrcommon.h | 102 ++++++++++++ .../include/streamrnode.h | 61 +++---- .../include/streamrproxyclient.h | 145 +++++++--------- .../src/streamrproxyclient.cpp | 29 +++- .../test/end-to-end/PublishToTsServerTest.cpp | 46 +++--- .../test/integration/StreamrNodeTest.cpp | 156 +++++++++--------- .../integration/StreamrProxyClientTest.cpp | 111 ++++++++----- .../ts-interop/StreamrNodeTsInteropTest.cpp | 26 +-- .../cpp/include/StreamrProxyClient.hpp | 28 ++-- .../libstreamrproxyclient.py | 20 +-- 11 files changed, 425 insertions(+), 303 deletions(-) create mode 100644 packages/streamr-libstreamrproxyclient/include/streamrcommon.h diff --git a/packages/streamr-libstreamrproxyclient/VERSION.cmake b/packages/streamr-libstreamrproxyclient/VERSION.cmake index 1d01dc02..8e0169bb 100644 --- a/packages/streamr-libstreamrproxyclient/VERSION.cmake +++ b/packages/streamr-libstreamrproxyclient/VERSION.cmake @@ -1,6 +1,6 @@ set(STREAMRPROXYCLIENT_VERSION -2.0.5 +3.0.0 ) set(STREAMRPROXYCLIENT_SOVERSION -2 +3 ) diff --git a/packages/streamr-libstreamrproxyclient/include/streamrcommon.h b/packages/streamr-libstreamrproxyclient/include/streamrcommon.h new file mode 100644 index 00000000..c55dcf18 --- /dev/null +++ b/packages/streamr-libstreamrproxyclient/include/streamrcommon.h @@ -0,0 +1,102 @@ +#ifndef STREAMR_COMMON_H +#define STREAMR_COMMON_H + +// Shared infrastructure of the Streamr shared-library C APIs: the result +// and error types, the peer locator struct, and the library lifecycle. +// Used by both the full-node API (streamrnode.h) and the proxy-client API +// (streamrproxyclient.h). These are the FINAL, API-neutral names decided +// in phase D3 of trackerless-network-completion-plan.md; the former +// proxy-prefixed names remain available as deprecated aliases in +// streamrproxyclient.h for one release. + +// NOLINTNEXTLINE +#include + +#if defined(__clang__) || defined(__GNUC__) +#define SHARED_EXPORT __attribute__((visibility("default"))) +#define SHARED_LOCAL __attribute__((visibility("hidden"))) +#define STREAMR_DEPRECATED(msg) __attribute__((deprecated(msg))) +#else +#define SHARED_EXPORT +#define SHARED_LOCAL +#define STREAMR_DEPRECATED(msg) +#endif + +#ifdef __cplusplus +#define EXTERN_C extern "C" +#else +#define EXTERN_C +#endif + +// Error codes shared by both APIs. +#define ERROR_INVALID_ETHEREUM_ADDRESS "INVALID_ETHEREUM_ADDRESS" +#define ERROR_INVALID_STREAM_PART_ID "INVALID_STREAM_PART_ID" + +/** + * @brief A peer locator: the websocket URL a node listens on plus the + * ethereum address its node id is derived from. Used both for the proxies + * of the proxy-client API and for the entry points of the full-node API. + */ +// NOLINTNEXTLINE +typedef struct StreamrPeer { + const char* websocketUrl; + const char* ethereumAddress; +} StreamrPeer; + +/** + * @brief One error of a failed or partially failed call. code is one of + * the ERROR_* constants; peer points at the peer the error concerns, or + * NULL when the error is not peer-specific. + */ +// NOLINTNEXTLINE +typedef struct StreamrError { + const char* message; + const char* code; + // The peer this error concerns. The member is accessible both as + // `peer` (final name) and as `proxy` (the pre-3.0 spelling, kept for + // source compatibility with proxy-API consumers; same storage). + union { + const struct StreamrPeer* peer; + const struct StreamrPeer* proxy; + }; +} StreamrError; + +/** + * @brief The result of every call in both APIs, returned through a + * const StreamrResult** out parameter. Every returned result — success or + * error — MUST be freed with streamrResultDelete(). Calls that cannot + * partially succeed leave `successful`/`numSuccessful` unused. + */ +// NOLINTNEXTLINE +typedef struct StreamrResult { + StreamrError* errors; + uint64_t numErrors; + StreamrPeer* successful; + uint64_t numSuccessful; +} StreamrResult; + +/** + * @brief Delete a StreamrResult. Must be called on every result returned + * by any call of either API. + */ +EXTERN_C SHARED_EXPORT void streamrResultDelete( + const StreamrResult* streamrResult); + +/** + * @brief Initialize the library. Called automatically when the library is + * loaded, but can be called explicitly to force a re-initialization (e.g. + * after streamrCleanupLibrary()). + */ +EXTERN_C SHARED_EXPORT void streamrInitLibrary(void); + +/** + * @brief Clean up the library. MUST be called before the program exits: + * the standard dynamic-library destructor runs after static variables + * have already been destroyed, too late to tear down the objects that + * depend on them. Safe to call multiple times. The library may be used + * again after a cleanup: any API call re-initializes it automatically + * (an explicit streamrInitLibrary() call is also fine). + */ +EXTERN_C SHARED_EXPORT void streamrCleanupLibrary(void); + +#endif /* STREAMR_COMMON_H */ diff --git a/packages/streamr-libstreamrproxyclient/include/streamrnode.h b/packages/streamr-libstreamrproxyclient/include/streamrnode.h index 45fc3a22..cd5dfebd 100644 --- a/packages/streamr-libstreamrproxyclient/include/streamrnode.h +++ b/packages/streamr-libstreamrproxyclient/include/streamrnode.h @@ -5,26 +5,25 @@ // trackerless-network-completion-plan.md). // // This API lives in the same shared library as the proxy-client API -// (streamrproxyclient.h) and reuses its infrastructure on purpose: -// - Results are reported through the same const ProxyResult** out +// (streamrproxyclient.h) and shares its infrastructure (streamrcommon.h): +// - Results are reported through the same const StreamrResult** out // parameter in every call; every returned result (success or error) -// MUST be freed with proxyClientResultDelete(). -// - The library lifecycle functions proxyClientInitLibrary() / -// proxyClientCleanupLibrary() govern this API too; -// proxyClientCleanupLibrary() stops and deletes any nodes that are +// MUST be freed with streamrResultDelete(). +// - The library lifecycle functions streamrInitLibrary() / +// streamrCleanupLibrary() govern this API too; +// streamrCleanupLibrary() stops and deletes any nodes that are // still alive. As with the proxy API, it MUST be called before the -// process exits (see the note in streamrproxyclient.h). -// - Entry points are (websocketUrl, ethereumAddress) pairs with the -// same layout as struct Proxy; the node id of an entry point is -// derived from its ethereum address exactly like the proxy API -// derives peer ids, so any streamrNode* node can act as an entry -// point for another one. -// NAMING: whether the shared types/lifecycle should be renamed to a -// streamr* prefix (a breaking change for proxy-API users) is left as a -// review decision for this PR. +// process exits (see the note in streamrcommon.h). +// - Entry points are (websocketUrl, ethereumAddress) pairs — struct +// StreamrPeer; the node id of an entry point is derived from its +// ethereum address exactly like the proxy API derives peer ids, so +// any streamrNode* node can act as an entry point for another one. +// (The shared-infrastructure naming decision deferred from the D1 review +// is resolved in phase D3a: neutral streamr* names in streamrcommon.h, +// proxy-prefixed aliases deprecated for one release.) #include // NOLINT -#include "streamrproxyclient.h" +#include "streamrcommon.h" #define ERROR_NODE_NOT_FOUND "NODE_NOT_FOUND" #define ERROR_NODE_NOT_STARTED "NODE_NOT_STARTED" @@ -37,7 +36,7 @@ // An entry point has the same shape as a proxy: the websocket URL the // node listens on plus the ethereum address its node id is derived // from. -typedef Proxy StreamrEntryPoint; +typedef StreamrPeer StreamrEntryPoint; typedef struct StreamrNodeConfig { // Entry points of the layer-0 DHT. May be NULL when numEntryPoints @@ -87,26 +86,26 @@ typedef void (*StreamrNodeMessageCallback)( // config may be NULL, which is equivalent to a zeroed config (no entry // points, no websocket server). Returns the node handle, or 0 on error. EXTERN_C SHARED_EXPORT uint64_t streamrNodeNew( - const ProxyResult** result, + const StreamrResult** result, const char* ownEthereumAddress, const StreamrNodeConfig* config); // Stops the node if it is still running and releases it. The handle is // invalid afterwards. EXTERN_C SHARED_EXPORT void streamrNodeDelete( - const ProxyResult** result, uint64_t nodeHandle); + const StreamrResult** result, uint64_t nodeHandle); // Starts the node: opens the websocket server (if configured) and // joins the layer-0 DHT through the configured entry points. Blocks // until the node is connected to the network. A node cannot be started // twice, and a stopped node cannot be restarted. EXTERN_C SHARED_EXPORT void streamrNodeStart( - const ProxyResult** result, uint64_t nodeHandle); + const StreamrResult** result, uint64_t nodeHandle); // Stops the node and leaves the network. Terminal: the node cannot be // restarted (create a new one instead). Idempotent. EXTERN_C SHARED_EXPORT void streamrNodeStop( - const ProxyResult** result, uint64_t nodeHandle); + const StreamrResult** result, uint64_t nodeHandle); // Joins a stream part. streamPartEntryPoints optionally names nodes // known to be in the stream part (same convention as the layer-0 entry @@ -114,7 +113,7 @@ EXTERN_C SHARED_EXPORT void streamrNodeStop( // discovered through the DHT. Returns once the join has been initiated; // use streamrNodeGetNeighborCount to observe the topology forming. EXTERN_C SHARED_EXPORT void streamrNodeJoinStreamPart( - const ProxyResult** result, + const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const StreamrEntryPoint* streamPartEntryPoints, @@ -122,7 +121,9 @@ EXTERN_C SHARED_EXPORT void streamrNodeJoinStreamPart( // Leaves a stream part. EXTERN_C SHARED_EXPORT void streamrNodeLeaveStreamPart( - const ProxyResult** result, uint64_t nodeHandle, const char* streamPartId); + const StreamrResult** result, + uint64_t nodeHandle, + const char* streamPartId); // Publishes a message to a stream part (joining it first if needed). // content is contentLength bytes of opaque payload. ethereumPrivateKey @@ -131,7 +132,7 @@ EXTERN_C SHARED_EXPORT void streamrNodeLeaveStreamPart( // proxyClientPublish. Fire-and-forget: delivery into the overlay is // not acknowledged. EXTERN_C SHARED_EXPORT void streamrNodePublish( - const ProxyResult** result, + const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const char* content, @@ -142,7 +143,7 @@ EXTERN_C SHARED_EXPORT void streamrNodePublish( // invoked for every message received on it. Returns a subscription // handle for streamrNodeUnsubscribe, or 0 on error. EXTERN_C SHARED_EXPORT uint64_t streamrNodeSubscribe( - const ProxyResult** result, + const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, StreamrNodeMessageCallback callback, @@ -151,14 +152,16 @@ EXTERN_C SHARED_EXPORT uint64_t streamrNodeSubscribe( // Removes a subscription. The callback may still be invoked for // messages whose dispatch already started when this returns. EXTERN_C SHARED_EXPORT void streamrNodeUnsubscribe( - const ProxyResult** result, + const StreamrResult** result, uint64_t nodeHandle, uint64_t subscriptionHandle); // Number of neighbors the node currently has in a stream part's // delivery topology. EXTERN_C SHARED_EXPORT uint64_t streamrNodeGetNeighborCount( - const ProxyResult** result, uint64_t nodeHandle, const char* streamPartId); + const StreamrResult** result, + uint64_t nodeHandle, + const char* streamPartId); // Connects the stream part through proxy nodes instead of joining its // topology (the proxy mode of the old proxy-client API folded into the @@ -166,10 +169,10 @@ EXTERN_C SHARED_EXPORT uint64_t streamrNodeGetNeighborCount( // the proxies; connectionCount 0 means "connect to all given proxies". // Passing numProxies 0 disconnects the stream part from its proxies. EXTERN_C SHARED_EXPORT void streamrNodeSetProxies( - const ProxyResult** result, + const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, - const Proxy* proxies, + const StreamrPeer* proxies, uint64_t numProxies, StreamrProxyDirection direction, uint64_t connectionCount); diff --git a/packages/streamr-libstreamrproxyclient/include/streamrproxyclient.h b/packages/streamr-libstreamrproxyclient/include/streamrproxyclient.h index d25573b7..5191d902 100644 --- a/packages/streamr-libstreamrproxyclient/include/streamrproxyclient.h +++ b/packages/streamr-libstreamrproxyclient/include/streamrproxyclient.h @@ -1,25 +1,15 @@ #ifndef LIBSTREAMRPROXYCLIENT_H #define LIBSTREAMRPROXYCLIENT_H -// NOLINTNEXTLINE -#include - -#if defined(__clang__) || defined(__GNUC__) -#define SHARED_EXPORT __attribute__((visibility("default"))) -#define SHARED_LOCAL __attribute__((visibility("hidden"))) -#else -#define SHARED_EXPORT -#define SHARED_LOCAL -#endif - -#ifdef __cplusplus -#define EXTERN_C extern "C" -#else -#define EXTERN_C -#endif - -#define ERROR_INVALID_ETHEREUM_ADDRESS "INVALID_ETHEREUM_ADDRESS" -#define ERROR_INVALID_STREAM_PART_ID "INVALID_STREAM_PART_ID" +// The proxy-client C API. The shared infrastructure (StreamrPeer, +// StreamrError, StreamrResult, streamrResultDelete, streamrInitLibrary, +// streamrCleanupLibrary) lives in streamrcommon.h and is shared with the +// full-node API (streamrnode.h); the pre-3.0 proxy-prefixed names below +// are deprecated aliases kept for one release. + +#include "streamrcommon.h" + +// Proxy-API-specific error codes (the shared ones are in streamrcommon.h). #define ERROR_PROXY_CLIENT_NOT_FOUND "PROXY_CLIENT_NOT_FOUND" #define ERROR_INVALID_PROXY_URL "INVALID_PROXY_URL" #define ERROR_NO_PROXIES_DEFINED "NO_PROXIES_DEFINED" @@ -31,63 +21,53 @@ EXTERN_C SHARED_EXPORT const char* testRpc(void); // NOLINTNEXTLINE SHARED_EXPORT static void __attribute__((constructor)) initialize(void); - -/** - * @brief Initialize the library. This function is called automatically when the library is loaded, - * but can be called explicitly to force a re-initialization (e.g. after proxyClientCleanupLibrary() has been called). - */ +// --------------------------------------------------------------------------- +// Deprecated pre-3.0 aliases of the shared infrastructure. New code (and +// the language wrappers) uses the streamrcommon.h names; these are kept +// for one release so existing proxy-API consumers keep compiling, and the +// old C symbols stay exported so existing binaries keep linking. +// --------------------------------------------------------------------------- // NOLINTNEXTLINE -EXTERN_C SHARED_EXPORT void proxyClientInitLibrary(void); - -/** - * @brief Cleanup the library. This function MUST be called before the program exits. - * Can be safely called multiple times. This is needed because the standard dynamic library - * destructor (__attribute__((destructor))) is called after static variables have already been destroyed, - * which makes it impossible to clean up the other objects that depend on the static variables. - * The library may be used again after a cleanup: any API call re-initializes it automatically - * (an explicit proxyClientInitLibrary() call is also fine). - */ +typedef StreamrPeer Proxy; // NOLINTNEXTLINE -EXTERN_C SHARED_EXPORT void proxyClientCleanupLibrary(void); - - +typedef StreamrError Error; // NOLINTNEXTLINE -typedef struct Proxy { - const char* websocketUrl; - const char* ethereumAddress; -} Proxy; +typedef StreamrResult ProxyResult; +/** + * @deprecated Use streamrInitLibrary(). + */ // NOLINTNEXTLINE -typedef struct Error { - const char* message; - const char* code; - const struct Proxy* proxy; -} Error; +EXTERN_C SHARED_EXPORT STREAMR_DEPRECATED( + "use streamrInitLibrary()") void proxyClientInitLibrary(void); +/** + * @deprecated Use streamrCleanupLibrary(). + */ // NOLINTNEXTLINE -typedef struct ProxyResult { - Error* errors; - uint64_t numErrors; - Proxy* successful; - uint64_t numSuccessful; -} ProxyResult; +EXTERN_C SHARED_EXPORT STREAMR_DEPRECATED( + "use streamrCleanupLibrary()") void proxyClientCleanupLibrary(void); /** - * @brief Delete a ProxyResult. This method must be called after every call that - * returns a ProxyResult. - * @param proxyResult The ProxyResult to delete. + * @deprecated Use streamrResultDelete(). */ +EXTERN_C SHARED_EXPORT + STREAMR_DEPRECATED("use streamrResultDelete()") void proxyClientResultDelete( + const StreamrResult* proxyResult); -EXTERN_C SHARED_EXPORT void proxyClientResultDelete(const ProxyResult* proxyResult); +// --------------------------------------------------------------------------- +// The proxy-client calls. These keep their names (the API itself is about +// proxies); only the shared types in their signatures changed spelling. +// --------------------------------------------------------------------------- /** * @brief Create a new proxy client. * - * @param proxyResult Pointer in which ProxyResult will be stored. You MUST call - * proxyClientResultDelete() on this after the call returns. The resulting ProxyResult - * may only contain "errors" - the "successful" and "numSuccessful" fields are - * unused. + * @param streamrResult Pointer in which the StreamrResult will be stored. + * You MUST call streamrResultDelete() on this after the call returns. The + * resulting StreamrResult may only contain "errors" — the "successful" + * and "numSuccessful" fields are unused. * @param ownEthereumAddress The Ethereum address of the client in format * 0x1234567890123456789012345678901234567890. * @param streamPartId The stream part id in format @@ -96,28 +76,28 @@ EXTERN_C SHARED_EXPORT void proxyClientResultDelete(const ProxyResult* proxyResu */ EXTERN_C SHARED_EXPORT uint64_t proxyClientNew( - const ProxyResult** proxyResult, + const StreamrResult** streamrResult, const char* ownEthereumAddress, const char* streamPartId); /** * @brief Delete a proxy client. * - * @param proxyResult Pointer in which ProxyResult will be stored. You MUST call - * proxyClientResultDelete() on this after the call returns. The resulting ProxyResult - * may only contain "errors" - the "successful" and "numSuccessful" fields are - * unused. + * @param streamrResult Pointer in which the StreamrResult will be stored. + * You MUST call streamrResultDelete() on this after the call returns. The + * resulting StreamrResult may only contain "errors" — the "successful" + * and "numSuccessful" fields are unused. * @param clientHandle The client handle of the client to delete. */ EXTERN_C SHARED_EXPORT void proxyClientDelete( - const ProxyResult** proxyResult, uint64_t clientHandle); + const StreamrResult** streamrResult, uint64_t clientHandle); /** * @brief Connect a proxy client to a list of proxies. * - * @param proxyResult Pointer in which ProxyResult will be stored. You MUST call - * proxyClientResultDelete() on this after the call returns. + * @param streamrResult Pointer in which the StreamrResult will be stored. + * You MUST call streamrResultDelete() on this after the call returns. * @param clientHandle The client handle of the client to connect. * @param proxies The array of proxies. * @param numProxies The number of proxies. @@ -125,39 +105,26 @@ EXTERN_C SHARED_EXPORT void proxyClientDelete( */ EXTERN_C SHARED_EXPORT uint64_t proxyClientConnect( - const ProxyResult** proxyResult, + const StreamrResult** streamrResult, uint64_t clientHandle, - const Proxy* proxies, + const StreamrPeer* proxies, uint64_t numProxies); -/** - * @brief Disconnect a proxy client from all proxies. - * - * @param proxyResult Pointer in which ProxyResult will be stored. You MUST call - * proxyClientResultDelete() on this after the call returns. The resulting ProxyResult - * may only contain "errors" - the "successful" and "numSuccessful" fields are - * unused. - * @param clientHandle The client handle of the client to disconnect. - -EXTERN_C SHARED_EXPORT void proxyClientDisconnect( - const ProxyResult** proxyResult, uint64_t clientHandle); -*/ - /** * @brief Publish a message to the stream. * - * @param proxyResult Pointer in which ProxyResult will be stored. You MUST call - * proxyClientResultDelete() on this after the call returns. + * @param streamrResult Pointer in which the StreamrResult will be stored. + * You MUST call streamrResultDelete() on this after the call returns. * @param clientHandle The client handle of the client to publish. * @param content The content to publish. * @param contentLength The length of the content. - * @param ethereumPrivateKey as hex without 0x prefix (64 characters) or NULL if - * message should not be signed. - * @return The number of proxies to which the message was published to. + * @param ethereumPrivateKey as hex without 0x prefix (64 characters) or + * NULL if the message should not be signed. + * @return The number of proxies to which the message was published. */ EXTERN_C SHARED_EXPORT uint64_t proxyClientPublish( - const ProxyResult** proxyResult, + const StreamrResult** streamrResult, uint64_t clientHandle, const char* content, uint64_t contentLength, diff --git a/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp b/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp index c9c589ee..5eb61955 100644 --- a/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp +++ b/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp @@ -29,6 +29,7 @@ #include #include #include "streamrnode.h" +#include "streamrproxyclient.h" import streamr.trackerlessnetwork.protos; import streamr.dht.ConnectionManager; @@ -1356,17 +1357,17 @@ static LibProxyClientApi* libProxyClientApi = nullptr; // NOLINT static void initialize() { // NOLINT // std::cout << "initialize()" << "\n"; - proxyClientInitLibrary(); + streamrInitLibrary(); } -void proxyClientInitLibrary() { // NOLINT +void streamrInitLibrary() { // NOLINT if (libProxyClientApi != nullptr) { return; } libProxyClientApi = new LibProxyClientApi(); } -void proxyClientCleanupLibrary() { // NOLINT +void streamrCleanupLibrary() { // NOLINT if (libProxyClientApi == nullptr) { return; } @@ -1384,9 +1385,27 @@ static LibProxyClientApi& getProxyClientApi() { // NOLINT return *libProxyClientApi; } -void proxyClientResultDelete(const ProxyResult* proxyResult) { - getProxyClientApi().proxyClientResultDelete(proxyResult); +void streamrResultDelete(const StreamrResult* streamrResult) { + getProxyClientApi().proxyClientResultDelete(streamrResult); +} + +// Deprecated pre-3.0 symbols, kept exported for one release (see +// streamrproxyclient.h). Defined here so existing binaries keep linking; +// the pragma silences the self-referential deprecation warnings. +#pragma clang diagnostic push +#pragma clang diagnostic ignored "-Wdeprecated-declarations" +void proxyClientInitLibrary() { // NOLINT + streamrInitLibrary(); +} + +void proxyClientCleanupLibrary() { // NOLINT + streamrCleanupLibrary(); +} + +void proxyClientResultDelete(const StreamrResult* proxyResult) { + streamrResultDelete(proxyResult); } +#pragma clang diagnostic pop uint64_t proxyClientNew( const ProxyResult** proxyResult, diff --git a/packages/streamr-libstreamrproxyclient/test/end-to-end/PublishToTsServerTest.cpp b/packages/streamr-libstreamrproxyclient/test/end-to-end/PublishToTsServerTest.cpp index 0ab77d04..51c161e7 100644 --- a/packages/streamr-libstreamrproxyclient/test/end-to-end/PublishToTsServerTest.cpp +++ b/packages/streamr-libstreamrproxyclient/test/end-to-end/PublishToTsServerTest.cpp @@ -14,77 +14,77 @@ class PublishToTsServerTest : public ::testing::Test { "0xa000000000000000000000000000000000000000#01"; public: - ~PublishToTsServerTest() override { proxyClientCleanupLibrary(); } + ~PublishToTsServerTest() override { streamrCleanupLibrary(); } }; TEST_F(PublishToTsServerTest, ProxyPublish) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t clientHandle = proxyClientNew(&result, ownEthereumAddress, tsStreamPartId); EXPECT_EQ(result->numErrors, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); - Proxy proxy{ + StreamrPeer proxy{ .websocketUrl = tsProxyUrl, .ethereumAddress = tsEthereumAddress}; - const ProxyResult* result2 = nullptr; + const StreamrResult* result2 = nullptr; proxyClientConnect(&result2, clientHandle, &proxy, 1); EXPECT_EQ(result2->numErrors, 0); - proxyClientResultDelete(result2); + streamrResultDelete(result2); std::string message = "Hello from libstreamrproxyclient!"; - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; proxyClientPublish( &result3, clientHandle, message.c_str(), message.length(), nullptr); EXPECT_EQ(result3->numErrors, 0); - proxyClientResultDelete(result3); + streamrResultDelete(result3); - const ProxyResult* result4 = nullptr; + const StreamrResult* result4 = nullptr; proxyClientDelete(&result4, clientHandle); - proxyClientResultDelete(result4); + streamrResultDelete(result4); } TEST_F(PublishToTsServerTest, ProxyPublishMultipleMessages) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t clientHandle = proxyClientNew(&result, ownEthereumAddress, tsStreamPartId); EXPECT_EQ(result->numErrors, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); - Proxy proxy{ + StreamrPeer proxy{ .websocketUrl = tsProxyUrl, .ethereumAddress = tsEthereumAddress}; - const ProxyResult* result2 = nullptr; + const StreamrResult* result2 = nullptr; proxyClientConnect(&result2, clientHandle, &proxy, 1); - proxyClientResultDelete(result2); + streamrResultDelete(result2); std::string message1 = "Hello from libstreamrproxyclient! m1"; std::string message2 = "Hello from libstreamrproxyclient! m2"; std::string message3 = "Hello from libstreamrproxyclient! m3"; - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; proxyClientPublish( &result3, clientHandle, message1.c_str(), message1.length(), nullptr); - proxyClientResultDelete(result3); + streamrResultDelete(result3); - const ProxyResult* result4 = nullptr; + const StreamrResult* result4 = nullptr; proxyClientPublish( &result4, clientHandle, message2.c_str(), message2.length(), nullptr); - proxyClientResultDelete(result4); + streamrResultDelete(result4); - const ProxyResult* result5 = nullptr; + const StreamrResult* result5 = nullptr; proxyClientPublish( &result5, clientHandle, message3.c_str(), message3.length(), nullptr); - proxyClientResultDelete(result5); + streamrResultDelete(result5); - const ProxyResult* result6 = nullptr; + const StreamrResult* result6 = nullptr; proxyClientDelete(&result6, clientHandle); - proxyClientResultDelete(result6); + streamrResultDelete(result6); } diff --git a/packages/streamr-libstreamrproxyclient/test/integration/StreamrNodeTest.cpp b/packages/streamr-libstreamrproxyclient/test/integration/StreamrNodeTest.cpp index 9d9da564..202252f9 100644 --- a/packages/streamr-libstreamrproxyclient/test/integration/StreamrNodeTest.cpp +++ b/packages/streamr-libstreamrproxyclient/test/integration/StreamrNodeTest.cpp @@ -82,13 +82,13 @@ class StreamrNodeTest : public ::testing::Test { // NOLINTNEXTLINE(bugprone-easily-swappable-parameters) static void expectSingleError( - const ProxyResult* result, const char* expectedCode) { + const StreamrResult* result, const char* expectedCode) { ASSERT_NE(result, nullptr); ASSERT_EQ(result->numErrors, 1); EXPECT_STREQ(result->errors[0].code, expectedCode); } - static void expectNoErrors(const ProxyResult* result) { + static void expectNoErrors(const StreamrResult* result) { ASSERT_NE(result, nullptr); EXPECT_EQ(result->numErrors, 0); } @@ -98,32 +98,32 @@ class StreamrNodeTest : public ::testing::Test { // library up after every test; re-initialize here so consecutive // tests also work when the binary is run as a single process (under // ctest each test is its own process and this is a no-op). - StreamrNodeTest() { proxyClientInitLibrary(); } - ~StreamrNodeTest() override { proxyClientCleanupLibrary(); } + StreamrNodeTest() { streamrInitLibrary(); } + ~StreamrNodeTest() override { streamrCleanupLibrary(); } }; TEST_F(StreamrNodeTest, InvalidEthereumAddress) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t nodeHandle = streamrNodeNew(&result, invalidEthereumAddress, nullptr); EXPECT_EQ(nodeHandle, 0); expectSingleError(result, ERROR_INVALID_ETHEREUM_ADDRESS); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, InvalidEntryPointUrl) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; StreamrEntryPoint entryPoint{ .websocketUrl = invalidUrl, .ethereumAddress = ethereumAddressB}; StreamrNodeConfig config{.entryPoints = &entryPoint, .numEntryPoints = 1}; uint64_t nodeHandle = streamrNodeNew(&result, ethereumAddressA, &config); EXPECT_EQ(nodeHandle, 0); expectSingleError(result, ERROR_INVALID_ENTRY_POINT_URL); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, InvalidEntryPointEthereumAddress) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; StreamrEntryPoint entryPoint{ .websocketUrl = "ws://127.0.0.1:40000", .ethereumAddress = invalidEthereumAddress}; @@ -131,34 +131,34 @@ TEST_F(StreamrNodeTest, InvalidEntryPointEthereumAddress) { uint64_t nodeHandle = streamrNodeNew(&result, ethereumAddressA, &config); EXPECT_EQ(nodeHandle, 0); expectSingleError(result, ERROR_INVALID_ETHEREUM_ADDRESS); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, NodeNotFound) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; streamrNodeStart(&result, nonExistentNodeHandle); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nonExistentNodeHandle); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeJoinStreamPart( &result, nonExistentNodeHandle, validStreamPartId, nullptr, 0); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeLeaveStreamPart( &result, nonExistentNodeHandle, validStreamPartId); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodePublish( &result, nonExistentNodeHandle, validStreamPartId, "x", 1, nullptr); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); uint64_t subscriptionHandle = streamrNodeSubscribe( &result, @@ -168,18 +168,18 @@ TEST_F(StreamrNodeTest, NodeNotFound) { nullptr); EXPECT_EQ(subscriptionHandle, 0); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeUnsubscribe(&result, nonExistentNodeHandle, 1); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeGetNeighborCount( &result, nonExistentNodeHandle, validStreamPartId); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); - Proxy proxy{ + StreamrPeer proxy{ .websocketUrl = "ws://127.0.0.1:40000", .ethereumAddress = ethereumAddressB}; streamrNodeSetProxies( @@ -191,36 +191,36 @@ TEST_F(StreamrNodeTest, NodeNotFound) { STREAMR_PROXY_DIRECTION_PUBLISH, 1); expectSingleError(result, ERROR_NODE_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, CanCreateAndDeleteWithoutStarting) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t nodeHandle = streamrNodeNew(&result, ethereumAddressA, nullptr); EXPECT_NE(nodeHandle, 0); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); - const ProxyResult* result2 = nullptr; + const StreamrResult* result2 = nullptr; streamrNodeDelete(&result2, nodeHandle); expectNoErrors(result2); - proxyClientResultDelete(result2); + streamrResultDelete(result2); } TEST_F(StreamrNodeTest, OperationsRequireStart) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t nodeHandle = streamrNodeNew(&result, ethereumAddressA, nullptr); ASSERT_NE(nodeHandle, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeJoinStreamPart( &result, nodeHandle, validStreamPartId, nullptr, 0); expectSingleError(result, ERROR_NODE_NOT_STARTED); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodePublish(&result, nodeHandle, validStreamPartId, "x", 1, nullptr); expectSingleError(result, ERROR_NODE_NOT_STARTED); - proxyClientResultDelete(result); + streamrResultDelete(result); uint64_t subscriptionHandle = streamrNodeSubscribe( &result, @@ -230,82 +230,82 @@ TEST_F(StreamrNodeTest, OperationsRequireStart) { nullptr); EXPECT_EQ(subscriptionHandle, 0); expectSingleError(result, ERROR_NODE_NOT_STARTED); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeGetNeighborCount(&result, nodeHandle, validStreamPartId); expectSingleError(result, ERROR_NODE_NOT_STARTED); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nodeHandle); expectSingleError(result, ERROR_NODE_NOT_STARTED); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeUnsubscribe(&result, nodeHandle, 1); expectSingleError(result, ERROR_SUBSCRIPTION_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeHandle); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, StartStopLifecycle) { // No entry points and no websocket server: the node starts an // isolated network of its own (it is its own entry point). - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t nodeHandle = streamrNodeNew(&result, ethereumAddressA, nullptr); ASSERT_NE(nodeHandle, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeHandle); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeHandle); expectSingleError(result, ERROR_NODE_ALREADY_STARTED); - proxyClientResultDelete(result); + streamrResultDelete(result); // The stream-part-id validation needs a running node because the // running-state check comes first. streamrNodeJoinStreamPart( &result, nodeHandle, invalidStreamPartId, nullptr, 0); expectSingleError(result, ERROR_INVALID_STREAM_PART_ID); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nodeHandle); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); // Stopping is idempotent. streamrNodeStop(&result, nodeHandle); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); // A stopped node cannot be restarted or used. streamrNodeStart(&result, nodeHandle); expectSingleError(result, ERROR_NODE_STOPPED); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodePublish(&result, nodeHandle, validStreamPartId, "x", 1, nullptr); expectSingleError(result, ERROR_NODE_STOPPED); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeHandle); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, TwoNodesExchangeMessages) { // Node A runs a websocket server and acts as the entry point of a // new network; node B joins through it as a websocket client. - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; StreamrNodeConfig configA{.websocketPort = exchangeEntryPointPort}; uint64_t nodeA = streamrNodeNew(&result, ethereumAddressA, &configA); ASSERT_NE(nodeA, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeA); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); const std::string entryPointUrl = "ws://127.0.0.1:" + std::to_string(exchangeEntryPointPort); @@ -315,10 +315,10 @@ TEST_F(StreamrNodeTest, TwoNodesExchangeMessages) { StreamrNodeConfig configB{.entryPoints = &entryPoint, .numEntryPoints = 1}; uint64_t nodeB = streamrNodeNew(&result, ethereumAddressB, &configB); ASSERT_NE(nodeB, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeB); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); // Both nodes subscribe (which also joins the stream part). ReceivedMessages receivedA; @@ -330,7 +330,7 @@ TEST_F(StreamrNodeTest, TwoNodesExchangeMessages) { ReceivedMessages::callback, &receivedA); ASSERT_NE(subscriptionA, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); uint64_t subscriptionB = streamrNodeSubscribe( &result, nodeB, @@ -338,18 +338,18 @@ TEST_F(StreamrNodeTest, TwoNodesExchangeMessages) { ReceivedMessages::callback, &receivedB); ASSERT_NE(subscriptionB, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); // Wait for the stream part topology to form between the two nodes. EXPECT_TRUE(waitUntil( [&]() { - const ProxyResult* pollResult = nullptr; + const StreamrResult* pollResult = nullptr; auto neighborsA = streamrNodeGetNeighborCount( &pollResult, nodeA, validStreamPartId); - proxyClientResultDelete(pollResult); + streamrResultDelete(pollResult); auto neighborsB = streamrNodeGetNeighborCount( &pollResult, nodeB, validStreamPartId); - proxyClientResultDelete(pollResult); + streamrResultDelete(pollResult); return neighborsA >= 1 && neighborsB >= 1; }, topologyTimeout)); @@ -364,7 +364,7 @@ TEST_F(StreamrNodeTest, TwoNodesExchangeMessages) { messageFromA.size(), ethereumPrivateKey); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); EXPECT_TRUE(waitUntil( [&]() { return receivedB.contains(messageFromA); }, messageTimeout)); @@ -377,30 +377,30 @@ TEST_F(StreamrNodeTest, TwoNodesExchangeMessages) { messageFromB.size(), nullptr); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); EXPECT_TRUE(waitUntil( [&]() { return receivedA.contains(messageFromB); }, messageTimeout)); streamrNodeUnsubscribe(&result, nodeB, subscriptionB); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeUnsubscribe(&result, nodeB, subscriptionB); expectSingleError(result, ERROR_SUBSCRIPTION_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeUnsubscribe(&result, nodeA, subscriptionA); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nodeB); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nodeA); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeB); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeA); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrNodeTest, ProxyModePublishesToAcceptingNode) { @@ -408,15 +408,15 @@ TEST_F(StreamrNodeTest, ProxyModePublishesToAcceptingNode) { // client-only node (own isolated DHT, no websocket server) that // proxy-publishes into A — the old proxy-client use case through // the node handle. - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; StreamrNodeConfig configA{ .websocketPort = proxyServerPort, .acceptProxyConnections = true}; uint64_t nodeA = streamrNodeNew(&result, ethereumAddressA, &configA); ASSERT_NE(nodeA, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeA); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); ReceivedMessages receivedA; uint64_t subscriptionA = streamrNodeSubscribe( @@ -426,12 +426,12 @@ TEST_F(StreamrNodeTest, ProxyModePublishesToAcceptingNode) { ReceivedMessages::callback, &receivedA); ASSERT_NE(subscriptionA, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); // A node that accepts proxy connections cannot itself use proxies. const std::string proxyUrl = "ws://127.0.0.1:" + std::to_string(proxyServerPort); - Proxy proxy{ + StreamrPeer proxy{ .websocketUrl = proxyUrl.c_str(), .ethereumAddress = ethereumAddressA}; streamrNodeSetProxies( &result, @@ -442,14 +442,14 @@ TEST_F(StreamrNodeTest, ProxyModePublishesToAcceptingNode) { STREAMR_PROXY_DIRECTION_PUBLISH, 1); expectSingleError(result, ERROR_NODE_OPERATION_FAILED); - proxyClientResultDelete(result); + streamrResultDelete(result); uint64_t nodeB = streamrNodeNew(&result, ethereumAddressB, nullptr); ASSERT_NE(nodeB, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeB); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeSetProxies( &result, @@ -460,12 +460,12 @@ TEST_F(StreamrNodeTest, ProxyModePublishesToAcceptingNode) { STREAMR_PROXY_DIRECTION_PUBLISH, 1); expectNoErrors(result); - proxyClientResultDelete(result); + streamrResultDelete(result); const std::string messageViaProxy = "hello via proxy"; EXPECT_TRUE(waitUntil( [&]() { - const ProxyResult* publishResult = nullptr; + const StreamrResult* publishResult = nullptr; streamrNodePublish( &publishResult, nodeB, @@ -473,17 +473,17 @@ TEST_F(StreamrNodeTest, ProxyModePublishesToAcceptingNode) { messageViaProxy.data(), messageViaProxy.size(), nullptr); - proxyClientResultDelete(publishResult); + streamrResultDelete(publishResult); return receivedA.contains(messageViaProxy); }, messageTimeout)); streamrNodeStop(&result, nodeB); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nodeA); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeB); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeA); - proxyClientResultDelete(result); + streamrResultDelete(result); } diff --git a/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp b/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp index 8da3fedf..e8c08e4a 100644 --- a/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp +++ b/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp @@ -32,7 +32,7 @@ class StreamrProxyClientTest : public ::testing::Test { static constexpr uint64_t invalidClientHandle = 0; public: - ~StreamrProxyClientTest() override { proxyClientCleanupLibrary(); } + ~StreamrProxyClientTest() override { streamrCleanupLibrary(); } }; TEST_F(StreamrProxyClientTest, CanCallApi) { @@ -40,7 +40,7 @@ TEST_F(StreamrProxyClientTest, CanCallApi) { } TEST_F(StreamrProxyClientTest, CanCreateAndDeleteProxyClient) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = validEthereumAddress; const char* streamPartId = validStreamPartId; @@ -50,13 +50,13 @@ TEST_F(StreamrProxyClientTest, CanCreateAndDeleteProxyClient) { EXPECT_EQ(result->numErrors, 0); EXPECT_NE(clientHandle, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); - const ProxyResult* result2 = nullptr; + const StreamrResult* result2 = nullptr; proxyClientDelete(&result2, clientHandle); SLogger::info("proxyClientDelete called"); EXPECT_EQ(result2->numErrors, 0); - proxyClientResultDelete(result2); + streamrResultDelete(result2); SLogger::info("test finished"); } @@ -103,7 +103,7 @@ TEST_F(StreamrProxyClientTest, ApiIsUsableAfterCleanupLibrary) { } TEST_F(StreamrProxyClientTest, InvalidEthereumAddress) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = invalidEthereumAddress; const char* streamPartId = validStreamPartId; @@ -117,11 +117,11 @@ TEST_F(StreamrProxyClientTest, InvalidEthereumAddress) { EXPECT_EQ(result->numErrors, 1); EXPECT_STREQ(result->errors->code, ERROR_INVALID_ETHEREUM_ADDRESS); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrProxyClientTest, InvalidStreamPartId) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = validEthereumAddress; const char* streamPartId = invalidStreamPartId; @@ -135,13 +135,13 @@ TEST_F(StreamrProxyClientTest, InvalidStreamPartId) { EXPECT_EQ(result->numErrors, 1); EXPECT_STREQ(result->errors->code, ERROR_INVALID_STREAM_PART_ID); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrProxyClientTest, ProxyClientNotFound) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; - Proxy proxy{ + StreamrPeer proxy{ .websocketUrl = invalidProxyUrl, .ethereumAddress = validEthereumAddress}; @@ -149,11 +149,11 @@ TEST_F(StreamrProxyClientTest, ProxyClientNotFound) { EXPECT_EQ(result->numErrors, 1); EXPECT_STREQ(result->errors->code, ERROR_PROXY_CLIENT_NOT_FOUND); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrProxyClientTest, NoProxiesDefined) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = validEthereumAddress; const char* streamPartId = validStreamPartId; @@ -165,11 +165,11 @@ TEST_F(StreamrProxyClientTest, NoProxiesDefined) { EXPECT_EQ(result->numErrors, 1); EXPECT_STREQ(result->errors->code, ERROR_NO_PROXIES_DEFINED); - proxyClientResultDelete(result); + streamrResultDelete(result); } TEST_F(StreamrProxyClientTest, InvalidProxyUrl) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = validEthereumAddress; const char* streamPartId = validStreamPartId; @@ -177,25 +177,25 @@ TEST_F(StreamrProxyClientTest, InvalidProxyUrl) { uint64_t clientHandle = proxyClientNew(&result, ownEthereumAddress, streamPartId); - proxyClientResultDelete(result); + streamrResultDelete(result); - const ProxyResult* result2 = nullptr; - Proxy proxy{ + const StreamrResult* result2 = nullptr; + StreamrPeer proxy{ .websocketUrl = invalidProxyUrl, .ethereumAddress = validEthereumAddress}; proxyClientConnect(&result2, clientHandle, &proxy, 1); EXPECT_EQ(result2->numErrors, 1); EXPECT_STREQ(result2->errors->code, ERROR_INVALID_PROXY_URL); - proxyClientResultDelete(result2); + streamrResultDelete(result2); - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; proxyClientDelete(&result3, clientHandle); - proxyClientResultDelete(result3); + streamrResultDelete(result3); } TEST_F(StreamrProxyClientTest, InvalidProxyEthereumAddress) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = validEthereumAddress; const char* streamPartId = validStreamPartId; @@ -203,23 +203,23 @@ TEST_F(StreamrProxyClientTest, InvalidProxyEthereumAddress) { uint64_t clientHandle = proxyClientNew(&result, ownEthereumAddress, streamPartId); - const ProxyResult* result2 = nullptr; - Proxy proxy{ + const StreamrResult* result2 = nullptr; + StreamrPeer proxy{ .websocketUrl = validProxyUrl, .ethereumAddress = invalidEthereumAddress}; proxyClientConnect(&result2, clientHandle, &proxy, 1); EXPECT_EQ(result2->numErrors, 1); EXPECT_STREQ(result2->errors->code, ERROR_INVALID_ETHEREUM_ADDRESS); - proxyClientResultDelete(result2); + streamrResultDelete(result2); - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; proxyClientDelete(&result3, clientHandle); - proxyClientResultDelete(result3); + streamrResultDelete(result3); } TEST_F(StreamrProxyClientTest, ProxyConnectionFailed) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = validEthereumAddress; const char* streamPartId = validStreamPartId; @@ -227,9 +227,9 @@ TEST_F(StreamrProxyClientTest, ProxyConnectionFailed) { uint64_t clientHandle = proxyClientNew(&result, ownEthereumAddress, streamPartId); - proxyClientResultDelete(result); - const ProxyResult* result2 = nullptr; - Proxy proxy{ + streamrResultDelete(result); + const StreamrResult* result2 = nullptr; + StreamrPeer proxy{ .websocketUrl = nonExistentProxyUrl0, .ethereumAddress = validEthereumAddress}; proxyClientConnect(&result2, clientHandle, &proxy, 1); @@ -240,15 +240,15 @@ TEST_F(StreamrProxyClientTest, ProxyConnectionFailed) { EXPECT_EQ(result2->numErrors, 1); EXPECT_STREQ(result2->errors->code, ERROR_PROXY_CONNECTION_FAILED); - proxyClientResultDelete(result2); + streamrResultDelete(result2); - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; proxyClientDelete(&result3, clientHandle); - proxyClientResultDelete(result3); + streamrResultDelete(result3); } TEST_F(StreamrProxyClientTest, ThreeProxyConnectionsFailed) { - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; const char* ownEthereumAddress = goodEthereumAddress; const char* streamPartId = validStreamPartId; @@ -256,10 +256,10 @@ TEST_F(StreamrProxyClientTest, ThreeProxyConnectionsFailed) { uint64_t clientHandle = proxyClientNew(&result, ownEthereumAddress, streamPartId); - proxyClientResultDelete(result); + streamrResultDelete(result); // NOLINTNEXTLINE - Proxy proxies[] = { + StreamrPeer proxies[] = { {.websocketUrl = nonExistentProxyUrl0, .ethereumAddress = validEthereumAddress}, {.websocketUrl = nonExistentProxyUrl1, @@ -267,7 +267,7 @@ TEST_F(StreamrProxyClientTest, ThreeProxyConnectionsFailed) { {.websocketUrl = nonExistentProxyUrl2, .ethereumAddress = validEthereumAddress3}}; - const ProxyResult* result2 = nullptr; + const StreamrResult* result2 = nullptr; auto numConnections = proxyClientConnect(&result2, clientHandle, proxies, 3); @@ -287,7 +287,38 @@ TEST_F(StreamrProxyClientTest, ThreeProxyConnectionsFailed) { SLogger::info("errors: " + std::string(result2->errors[i].message)); EXPECT_STREQ(result2->errors[i].code, ERROR_PROXY_CONNECTION_FAILED); } - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; proxyClientDelete(&result3, clientHandle); - proxyClientResultDelete(result3); + streamrResultDelete(result3); +} + +// The pre-3.0 proxy-prefixed names must stay source- and ABI-compatible +// for one release (phase D3a): this test compiles against the deprecated +// aliases and calls the deprecated symbols on purpose. +#pragma clang diagnostic push +#pragma clang diagnostic ignored "-Wdeprecated-declarations" +TEST(StreamrProxyClientDeprecatedAliasTest, Pre30NamesStillWork) { + proxyClientInitLibrary(); + const ProxyResult* result = nullptr; + Proxy proxy{ + .websocketUrl = "ws://127.0.0.1:44211", + .ethereumAddress = "0x1234567890123456789012345678901234567890"}; + (void)proxy; + const uint64_t handle = proxyClientNew( + &result, + "0xa5374e3c19f15e1847881979dd0c6c9ffe846bd5", + "0xd2078dc2d780029473a39ce873fc182587be69db/low-level-client#0"); + ASSERT_NE(result, nullptr); + EXPECT_EQ(result->numErrors, 0); + proxyClientResultDelete(result); + result = nullptr; + proxyClientDelete(&result, handle); + ASSERT_NE(result, nullptr); + // The Error alias and its pre-3.0 `proxy` member spelling stay usable. + if (result->numErrors > 0) { + const Error* firstError = &result->errors[0]; + (void)firstError->proxy; + } + proxyClientResultDelete(result); } +#pragma clang diagnostic pop diff --git a/packages/streamr-libstreamrproxyclient/test/ts-interop/StreamrNodeTsInteropTest.cpp b/packages/streamr-libstreamrproxyclient/test/ts-interop/StreamrNodeTsInteropTest.cpp index b10ef8bf..5c9da615 100644 --- a/packages/streamr-libstreamrproxyclient/test/ts-interop/StreamrNodeTsInteropTest.cpp +++ b/packages/streamr-libstreamrproxyclient/test/ts-interop/StreamrNodeTsInteropTest.cpp @@ -153,7 +153,7 @@ class StreamrNodeTsInteropTest : public ::testing::Test { std::set tsReceived; // texts the TS node reported void SetUp() override { - proxyClientInitLibrary(); + streamrInitLibrary(); const auto nodeBinary = findNodeBinary(); if (!nodeBinary) { GTEST_SKIP() << "No node runtime >= " << minNodeMajorVersion @@ -172,7 +172,7 @@ class StreamrNodeTsInteropTest : public ::testing::Test { void TearDown() override { stopDriver(); - proxyClientCleanupLibrary(); + streamrCleanupLibrary(); } static std::filesystem::path harnessDir() { @@ -354,23 +354,23 @@ TEST_F(StreamrNodeTsInteropTest, MessagesFlowBothWaysWithTsNetwork) { // A client-only node: no websocket server, exactly the mobile // configuration milestone D targets. - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; StreamrNodeConfig config{.entryPoints = &entryPoint, .numEntryPoints = 1}; const uint64_t nodeHandle = streamrNodeNew(&result, cppEthereumAddress, &config); ASSERT_NE(nodeHandle, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStart(&result, nodeHandle); ASSERT_EQ(result->numErrors, 0) << "start failed: " << result->errors[0].message; - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeJoinStreamPart( &result, nodeHandle, interopStreamPartId, &entryPoint, 1); ASSERT_EQ(result->numErrors, 0) << "join failed: " << result->errors[0].message; - proxyClientResultDelete(result); + streamrResultDelete(result); ReceivedMessages received; const uint64_t subscriptionHandle = streamrNodeSubscribe( @@ -380,16 +380,16 @@ TEST_F(StreamrNodeTsInteropTest, MessagesFlowBothWaysWithTsNetwork) { ReceivedMessages::callback, &received); ASSERT_NE(subscriptionHandle, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); // Wait until the C node and the TS node are stream-part neighbors. const auto joinDeadline = std::chrono::steady_clock::now() + joinTimeout; uint64_t neighborCount = 0; while (std::chrono::steady_clock::now() < joinDeadline) { - const ProxyResult* pollResult = nullptr; + const StreamrResult* pollResult = nullptr; neighborCount = streamrNodeGetNeighborCount( &pollResult, nodeHandle, interopStreamPartId); - proxyClientResultDelete(pollResult); + streamrResultDelete(pollResult); if (neighborCount >= 1) { break; } @@ -409,7 +409,7 @@ TEST_F(StreamrNodeTsInteropTest, MessagesFlowBothWaysWithTsNetwork) { nullptr); ASSERT_EQ(result->numErrors, 0) << "publish failed: " << result->errors[0].message; - proxyClientResultDelete(result); + streamrResultDelete(result); EXPECT_TRUE(this->waitForTsReceived(cppText, messageTimeout)) << "the TS node never reported the C API message"; @@ -426,10 +426,10 @@ TEST_F(StreamrNodeTsInteropTest, MessagesFlowBothWaysWithTsNetwork) { << "the C API callback never got the TS message"; streamrNodeUnsubscribe(&result, nodeHandle, subscriptionHandle); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeStop(&result, nodeHandle); EXPECT_EQ(result->numErrors, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); streamrNodeDelete(&result, nodeHandle); - proxyClientResultDelete(result); + streamrResultDelete(result); } diff --git a/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp b/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp index fd334677..d0c0e816 100644 --- a/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp +++ b/packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrProxyClient.hpp @@ -78,7 +78,7 @@ struct StreamrProxyAddress { std::string websocketUrl; /**< WebSocket URL of the proxy */ std::string ethereumAddress; /**< Ethereum address of the proxy */ - static StreamrProxyAddress fromCProxy(const Proxy* proxy) { + static StreamrProxyAddress fromCProxy(const StreamrPeer* proxy) { return StreamrProxyAddress{ .websocketUrl = std::string(proxy->websocketUrl), .ethereumAddress = std::string(proxy->ethereumAddress)}; @@ -93,7 +93,7 @@ struct StreamrProxyError : public std::runtime_error { StreamrProxyErrorCode code; /**< The error code */ // NOLINTNEXTLINE(google-explicit-constructor) - StreamrProxyError(const Error* error) + StreamrProxyError(const StreamrError* error) : std::runtime_error(error->message), code(getErrorCode(error->code)) { if (error->proxy) { proxy = StreamrProxyAddress::fromCProxy(error->proxy); @@ -137,7 +137,7 @@ struct StreamrProxyResult { errors; /**< List of failed proxy operations */ // NOLINTNEXTLINE(google-explicit-constructor) - StreamrProxyResult(const ProxyResult* result) { + StreamrProxyResult(const StreamrResult* result) { this->successful.reserve(result->numSuccessful); for (size_t i = 0; i < result->numSuccessful; i++) { this->successful.push_back( @@ -152,8 +152,8 @@ struct StreamrProxyResult { class SharedLibraryWrapper { public: - SharedLibraryWrapper() { proxyClientInitLibrary(); } - ~SharedLibraryWrapper() { proxyClientCleanupLibrary(); } + SharedLibraryWrapper() { streamrInitLibrary(); } + ~SharedLibraryWrapper() { streamrCleanupLibrary(); } }; /** @@ -179,22 +179,22 @@ class StreamrProxyClient { const std::string& ethereumPrivateKey, // NOLINT const std::string& streamPartId) : ethereumPrivateKey(ethereumPrivateKey) { - const ProxyResult* result; + const StreamrResult* result; this->proxyClientHandle = proxyClientNew( &result, ownEthereumAddress.c_str(), streamPartId.c_str()); if (result->numErrors > 0) { auto error = StreamrProxyError(&result->errors[0]); - proxyClientResultDelete(result); + streamrResultDelete(result); throw error; // NOLINT } - proxyClientResultDelete(result); + streamrResultDelete(result); } ~StreamrProxyClient() { if (this->proxyClientHandle != 0) { - const ProxyResult* result; + const StreamrResult* result; proxyClientDelete(&result, this->proxyClientHandle); - proxyClientResultDelete(result); + streamrResultDelete(result); } } @@ -205,7 +205,7 @@ class StreamrProxyClient { */ [[nodiscard]] StreamrProxyResult connect( const std::vector& proxies) const { - const ProxyResult* result; + const StreamrResult* result; std::vector cProxies; cProxies.reserve(proxies.size()); for (const auto& proxy : proxies) { @@ -216,7 +216,7 @@ class StreamrProxyClient { proxyClientConnect( &result, this->proxyClientHandle, cProxies.data(), cProxies.size()); StreamrProxyResult streamrProxyResult(result); - proxyClientResultDelete(result); + streamrResultDelete(result); return streamrProxyResult; } @@ -226,7 +226,7 @@ class StreamrProxyClient { * @return Result containing successful publications and errors */ [[nodiscard]] StreamrProxyResult publish(const std::string& message) const { - const ProxyResult* result; + const StreamrResult* result; proxyClientPublish( &result, this->proxyClientHandle, @@ -234,7 +234,7 @@ class StreamrProxyClient { message.size(), this->ethereumPrivateKey.c_str()); StreamrProxyResult streamrProxyResult(result); - proxyClientResultDelete(result); + streamrResultDelete(result); return streamrProxyResult; } }; diff --git a/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py b/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py index 36354a2d..03213b69 100644 --- a/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py +++ b/packages/streamr-libstreamrproxyclient/wrappers/python/src/streamrproxyclient/libstreamrproxyclient.py @@ -174,12 +174,12 @@ def __enter__(self): self.lib.testRpc.restype = ctypes.c_char_p - self.lib.proxyClientInitLibrary.restype = None + self.lib.streamrInitLibrary.restype = None - self.lib.proxyClientCleanupLibrary.restype = None + self.lib.streamrCleanupLibrary.restype = None - self.lib.proxyClientResultDelete.argtypes = [ctypes.POINTER(CProxyClientResult)] - self.lib.proxyClientResultDelete.restype = None + self.lib.streamrResultDelete.argtypes = [ctypes.POINTER(CProxyClientResult)] + self.lib.streamrResultDelete.restype = None self.lib.proxyClientNew.argtypes = [ctypes.POINTER(ctypes.POINTER(CProxyClientResult)), ctypes.c_char_p, ctypes.c_char_p] self.lib.proxyClientNew.restype = ctypes.c_uint64 @@ -193,7 +193,7 @@ def __enter__(self): self.lib.proxyClientPublish.argtypes = [ctypes.POINTER(ctypes.POINTER(CProxyClientResult)), ctypes.c_uint64, ctypes.c_char_p, ctypes.c_uint64, ctypes.c_char_p] self.lib.proxyClientPublish.restype = ctypes.c_uint64 - self.lib.proxyClientInitLibrary() + self.lib.streamrInitLibrary() return self @@ -201,7 +201,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): """ Cleanup the library. """ - self.lib.proxyClientCleanupLibrary() + self.lib.streamrCleanupLibrary() class ProxyClient: """ @@ -229,7 +229,7 @@ def __enter__(self): self.clientHandle = self.lib.proxyClientNew(ctypes.byref(result), self.ownEthereumAddress.encode('utf-8'), self.streamPartId.encode('utf-8')) if result.contents.numErrors > 0: raise ProxyClientException(Error(result.contents.errors[0])) - self.lib.proxyClientResultDelete(result) + self.lib.streamrResultDelete(result) return self def __exit__(self, exc_type, exc_val, exc_tb): @@ -239,7 +239,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): result = ctypes.POINTER(CProxyClientResult)() self.lib.proxyClientDelete(ctypes.byref(result), self.clientHandle) assert result.contents.numErrors == 0 - self.lib.proxyClientResultDelete(result) + self.lib.streamrResultDelete(result) def connect(self, proxies: list[Proxy]) -> ProxyClientResult: """ @@ -253,7 +253,7 @@ def connect(self, proxies: list[Proxy]) -> ProxyClientResult: result = ctypes.POINTER(CProxyClientResult)() self.lib.proxyClientConnect(ctypes.byref(result), self.clientHandle, proxy_array, num_proxies) res = ProxyClientResult(result) - self.lib.proxyClientResultDelete(result) + self.lib.streamrResultDelete(result) return res def publish(self, data: bytes, ethereumPrivateKey: str = None) -> ProxyClientResult: @@ -270,5 +270,5 @@ def publish(self, data: bytes, ethereumPrivateKey: str = None) -> ProxyClientRes else: self.lib.proxyClientPublish(ctypes.byref(result), self.clientHandle, data, len(data), None) res = ProxyClientResult(result) - self.lib.proxyClientResultDelete(result) + self.lib.streamrResultDelete(result) return res From c65035a667da4dbf2cc96dd1e744ffef6aa58d0d Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 18 Jul 2026 08:54:19 +0300 Subject: [PATCH 2/5] Phase D3b: C++ and Python wrappers for the streamrNode* full-node API (#96) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - wrappers/cpp: new StreamrNode.hpp — RAII node wrapper (create/start/ stop, joinStreamPart/leaveStreamPart, publish, subscribe with a std::function callback, neighborCount, setProxies), errors thrown as StreamrNodeError with a variant code, mirroring the existing StreamrProxyClient wrapper idioms. Subscription callbacks stay owned by the node until its destruction: the C API may still dispatch a message whose delivery already started when unsubscribe returns. - wrappers/python: StreamrNode context manager over the streamrNode* ctypes bindings (same shape as ProxyClient); the message callback is bound with an explicit content length (payloads are opaque bytes). ctypes callback objects are kept alive until node deletion. - Tests: StreamrNodeCppWrapperTest (two wrapper-created nodes form a stream-part topology and exchange a signed and an unsigned message; error mapping) wired into ctest with the streamrnode integration tests' 600 s budget, and the same round-trip in wrappers/python/tests/test_streamr_node.py with scripts/run-python-tests.sh staging the built library (pytest or direct run). - StreamrProxyClient.hpp: one leftover deprecated-alias use (std::vector) moved to the final StreamrPeer name. Local runs: cpp wrapper round-trip 2/2 in ~850 ms; python round-trip green against the freshly built 3.0.0 dylib. Swift (dist/ios-swift-package) and Kotlin (dist/android-library-module) follow in the next increment together with the D2 mobile-platform verification they depend on; the Go wrapper (wrappers/go, an uninitialized submodule pointing at a separate repo) is proposed for dropping from the D3 scope — decision left to review. Co-authored-by: Claude Fable 5 --- .../CMakeLists.txt | 22 +- .../scripts/run-python-tests.sh | 32 ++ .../src/streamrproxyclient.cpp | 4 +- .../wrappers/cpp/include/StreamrNode.hpp | 344 ++++++++++++++++++ .../cpp/include/StreamrProxyClient.hpp | 2 +- .../cpp/test/StreamrNodeCppWrapperTest.cpp | 134 +++++++ .../libstreamrproxyclient.py | 228 ++++++++++++ .../python/tests/test_streamr_node.py | 102 ++++++ 8 files changed, 861 insertions(+), 7 deletions(-) create mode 100755 packages/streamr-libstreamrproxyclient/scripts/run-python-tests.sh create mode 100644 packages/streamr-libstreamrproxyclient/wrappers/cpp/include/StreamrNode.hpp create mode 100644 packages/streamr-libstreamrproxyclient/wrappers/cpp/test/StreamrNodeCppWrapperTest.cpp create mode 100644 packages/streamr-libstreamrproxyclient/wrappers/python/tests/test_streamr_node.py 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/src/streamrproxyclient.cpp b/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp index 5eb61955..a7687de9 100644 --- a/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp +++ b/packages/streamr-libstreamrproxyclient/src/streamrproxyclient.cpp @@ -1376,12 +1376,12 @@ void streamrCleanupLibrary() { // NOLINT } static LibProxyClientApi& getProxyClientApi() { // NOLINT - // proxyClientCleanupLibrary() destroys the instance and only the + // streamrCleanupLibrary() destroys the instance and only the // load-time constructor re-creates it, so an API call made after a // cleanup would dereference a null pointer here. Re-initialize on // demand instead — the header documents that the library can be // re-initialized after a cleanup. - proxyClientInitLibrary(); + streamrInitLibrary(); return *libProxyClientApi; } 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") From c4b91c2a2f67ba1009db70c52abf6ce15a4855de Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 18 Jul 2026 12:39:27 +0300 Subject: [PATCH 3/5] Port the use-after-cleanup regression test to the streamr* names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebase fallout: #97 wrote ApiIsUsableAfterCleanupLibrary against the pre-3.0 proxyClient* names, which this branch deprecates — the lint step (deprecation warnings as errors) failed on every CI leg. The test verifies the re-init contract, not the aliases, so it now uses the final names; the dedicated deprecated-alias test keeps covering the old ones. Co-Authored-By: Claude Fable 5 --- .../integration/StreamrProxyClientTest.cpp | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp b/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp index e8c08e4a..96b8cd5d 100644 --- a/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp +++ b/packages/streamr-libstreamrproxyclient/test/integration/StreamrProxyClientTest.cpp @@ -62,7 +62,7 @@ TEST_F(StreamrProxyClientTest, CanCreateAndDeleteProxyClient) { } // Regression test for use-after-cleanup: the fixture destructor calls -// proxyClientCleanupLibrary() after every test, so any test that runs +// streamrCleanupLibrary() after every test, so any test that runs // after another one in the same process starts with the library torn // down. API entry points must re-initialize the library on demand // instead of dereferencing the destroyed instance (this crashed with @@ -70,36 +70,36 @@ TEST_F(StreamrProxyClientTest, CanCreateAndDeleteProxyClient) { // own process). TEST_F(StreamrProxyClientTest, ApiIsUsableAfterCleanupLibrary) { // Implicit re-initialization: call the API right after a cleanup. - proxyClientCleanupLibrary(); + streamrCleanupLibrary(); - const ProxyResult* result = nullptr; + const StreamrResult* result = nullptr; uint64_t clientHandle = proxyClientNew(&result, validEthereumAddress, validStreamPartId); EXPECT_EQ(result->numErrors, 0); EXPECT_NE(clientHandle, 0); - proxyClientResultDelete(result); + streamrResultDelete(result); - const ProxyResult* result2 = nullptr; + const StreamrResult* result2 = nullptr; proxyClientDelete(&result2, clientHandle); EXPECT_EQ(result2->numErrors, 0); - proxyClientResultDelete(result2); + streamrResultDelete(result2); // Explicit re-initialization (the path documented in - // streamrproxyclient.h): cleanup, init, use. - proxyClientCleanupLibrary(); - proxyClientInitLibrary(); + // streamrcommon.h): cleanup, init, use. + streamrCleanupLibrary(); + streamrInitLibrary(); - const ProxyResult* result3 = nullptr; + const StreamrResult* result3 = nullptr; uint64_t clientHandle2 = proxyClientNew(&result3, validEthereumAddress, validStreamPartId); EXPECT_EQ(result3->numErrors, 0); EXPECT_NE(clientHandle2, 0); - proxyClientResultDelete(result3); + streamrResultDelete(result3); - const ProxyResult* result4 = nullptr; + const StreamrResult* result4 = nullptr; proxyClientDelete(&result4, clientHandle2); EXPECT_EQ(result4->numErrors, 0); - proxyClientResultDelete(result4); + streamrResultDelete(result4); } TEST_F(StreamrProxyClientTest, InvalidEthereumAddress) { From 96ad2778add395ebc0f6dc2d605dfa0e2f598b0d Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 18 Jul 2026 12:47:52 +0300 Subject: [PATCH 4/5] CI: bump the build-tree cache generation to v2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The buildtree cache restores the newest previous tree on any key miss and rebuilds incrementally. The repeated force-pushes on this branch fed the legs an inconsistent restored state: ninja under-rebuilt, linking the freshly named libstreamrproxyclient from a stale pre-rename object (undefined streamrResultDelete/streamrInitLibrary in the wrapper test) and failing dyndep generation in streamr-proto-rpc. Bumping the generation forces one clean full build per leg and starts a fresh cache lineage — the same documented remedy the v1 suffix itself came from. Co-Authored-By: Claude Fable 5 --- .github/workflows/reusable/cached-install/action.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/reusable/cached-install/action.yml b/.github/workflows/reusable/cached-install/action.yml index 0bd53667..13969d4e 100644 --- a/.github/workflows/reusable/cached-install/action.yml +++ b/.github/workflows/reusable/cached-install/action.yml @@ -95,10 +95,10 @@ runs: id: cache-buildtree uses: actions/cache/restore@v4 with: - key: ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v1-${{ steps.keys.outputs.vcpkg_sha }}-${{ github.sha }} + key: ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v2-${{ steps.keys.outputs.vcpkg_sha }}-${{ github.sha }} restore-keys: | - ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v1-${{ steps.keys.outputs.vcpkg_sha }}- - ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v1- + ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v2-${{ steps.keys.outputs.vcpkg_sha }}- + ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v2- path: | ./build !./build/vcpkg_installed @@ -160,7 +160,7 @@ runs: if: github.ref == 'refs/heads/main' && steps.cache-buildtree.outputs.cache-hit != 'true' uses: actions/cache/save@v4 with: - key: ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v1-${{ steps.keys.outputs.vcpkg_sha }}-${{ github.sha }} + key: ${{ runner.arch }}-${{ runner.os }}-${{ steps.keys.outputs.platform }}-buildtree-v2-${{ steps.keys.outputs.vcpkg_sha }}-${{ github.sha }} path: | ./build !./build/vcpkg_installed From 70e73ccfbf673d1b871bf77a0d4a08f3053dd0a9 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 18 Jul 2026 13:04:35 +0300 Subject: [PATCH 5/5] Point the go wrapper submodule at the streamrNode* API branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit goproxyclient PR 1 (branch streamr-node-api, commit c500e43) adds the full-node API to the Go wrapper — the last wrapper of phase D3b's native-host set. The pointer references the pushed PR branch; advance it to the merge commit when that PR lands. Co-Authored-By: Claude Fable 5 --- packages/streamr-libstreamrproxyclient/wrappers/go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/streamr-libstreamrproxyclient/wrappers/go b/packages/streamr-libstreamrproxyclient/wrappers/go index 55f17ba3..c500e43d 160000 --- a/packages/streamr-libstreamrproxyclient/wrappers/go +++ b/packages/streamr-libstreamrproxyclient/wrappers/go @@ -1 +1 @@ -Subproject commit 55f17ba3850938aa7539bb0f22409b4abe1b15a4 +Subproject commit c500e43d95c74b9e873b5df1364d74985a7a82dc