diff --git a/dist/arm64-osx/lib/Release/libstreamrproxyclient.dylib b/dist/arm64-osx/lib/Release/libstreamrproxyclient.dylib index d7708f2..ebcfbc2 100755 Binary files a/dist/arm64-osx/lib/Release/libstreamrproxyclient.dylib and b/dist/arm64-osx/lib/Release/libstreamrproxyclient.dylib differ diff --git a/shim.c b/shim.c index 95824dd..9b1b308 100644 --- a/shim.c +++ b/shim.c @@ -1,6 +1,7 @@ #include #include #include "shim.h" +#include void (*proxyClientInitLibraryPtr)(void); // NOLINT void (*proxyClientCleanupLibraryPtr)(void); // NOLINT @@ -12,6 +13,18 @@ void (*proxyClientDisconnectPtr)(const ProxyResult**, uint64_t); uint64_t (*proxyClientPublishPtr)(const ProxyResult**, uint64_t, const char*, uint64_t, const char*); void* proxyClientLibrary; +uint64_t (*streamrNodeNewPtr)(const StreamrResult**, const char*, const StreamrNodeConfig*); +void (*streamrNodeDeletePtr)(const StreamrResult**, uint64_t); +void (*streamrNodeStartPtr)(const StreamrResult**, uint64_t); +void (*streamrNodeStopPtr)(const StreamrResult**, uint64_t); +void (*streamrNodeJoinStreamPartPtr)(const StreamrResult**, uint64_t, const char*, const StreamrPeer*, uint64_t); +void (*streamrNodeLeaveStreamPartPtr)(const StreamrResult**, uint64_t, const char*); +void (*streamrNodePublishPtr)(const StreamrResult**, uint64_t, const char*, const char*, uint64_t, const char*); +uint64_t (*streamrNodeSubscribePtr)(const StreamrResult**, uint64_t, const char*, StreamrNodeMessageCallback, void*); +void (*streamrNodeUnsubscribePtr)(const StreamrResult**, uint64_t, uint64_t); +uint64_t (*streamrNodeGetNeighborCountPtr)(const StreamrResult**, uint64_t, const char*); +void (*streamrNodeSetProxiesPtr)(const StreamrResult**, uint64_t, const char*, const StreamrPeer*, uint64_t, StreamrProxyDirection, uint64_t); + void loadSharedLibrary(const char* libFilePath) { proxyClientLibrary = dlopen(libFilePath, RTLD_LOCAL | RTLD_LAZY); proxyClientInitLibraryPtr = (void (*)(void))dlsym(proxyClientLibrary, "proxyClientInitLibrary"); // NOLINT @@ -22,6 +35,26 @@ void loadSharedLibrary(const char* libFilePath) { proxyClientConnectPtr = (uint64_t (*)(const ProxyResult**, uint64_t, const Proxy*, uint64_t))dlsym(proxyClientLibrary, "proxyClientConnect"); proxyClientDisconnectPtr = (void (*)(const ProxyResult**, uint64_t))dlsym(proxyClientLibrary, "proxyClientDisconnect"); proxyClientPublishPtr = (uint64_t (*)(const ProxyResult**, uint64_t, const char*, uint64_t, const char*))dlsym(proxyClientLibrary, "proxyClientPublish"); + streamrNodeNewPtr = (uint64_t (*)(const StreamrResult**, const char*, const StreamrNodeConfig*))dlsym(proxyClientLibrary, "streamrNodeNew"); + streamrNodeDeletePtr = (void (*)(const StreamrResult**, uint64_t))dlsym(proxyClientLibrary, "streamrNodeDelete"); + streamrNodeStartPtr = (void (*)(const StreamrResult**, uint64_t))dlsym(proxyClientLibrary, "streamrNodeStart"); + streamrNodeStopPtr = (void (*)(const StreamrResult**, uint64_t))dlsym(proxyClientLibrary, "streamrNodeStop"); + streamrNodeJoinStreamPartPtr = (void (*)(const StreamrResult**, uint64_t, const char*, const StreamrPeer*, uint64_t))dlsym(proxyClientLibrary, "streamrNodeJoinStreamPart"); + streamrNodeLeaveStreamPartPtr = (void (*)(const StreamrResult**, uint64_t, const char*))dlsym(proxyClientLibrary, "streamrNodeLeaveStreamPart"); + streamrNodePublishPtr = (void (*)(const StreamrResult**, uint64_t, const char*, const char*, uint64_t, const char*))dlsym(proxyClientLibrary, "streamrNodePublish"); + streamrNodeSubscribePtr = (uint64_t (*)(const StreamrResult**, uint64_t, const char*, StreamrNodeMessageCallback, void*))dlsym(proxyClientLibrary, "streamrNodeSubscribe"); + streamrNodeUnsubscribePtr = (void (*)(const StreamrResult**, uint64_t, uint64_t))dlsym(proxyClientLibrary, "streamrNodeUnsubscribe"); + streamrNodeGetNeighborCountPtr = (uint64_t (*)(const StreamrResult**, uint64_t, const char*))dlsym(proxyClientLibrary, "streamrNodeGetNeighborCount"); + streamrNodeSetProxiesPtr = (void (*)(const StreamrResult**, uint64_t, const char*, const StreamrPeer*, uint64_t, StreamrProxyDirection, uint64_t))dlsym(proxyClientLibrary, "streamrNodeSetProxies"); +} + +// Defined by cgo (//export goStreamrNodeMessage in streamrnode.go). +extern void goStreamrNodeMessage(uint64_t nodeHandle, char* streamPartId, char* content, uint64_t contentLength, void* userData); + +// C-ABI trampoline handed to streamrNodeSubscribe: bridges the const +// signature of the C API to the cgo export. +static void shimStreamrNodeMessageCallback(uint64_t nodeHandle, const char* streamPartId, const char* content, uint64_t contentLength, void* userData) { + goStreamrNodeMessage(nodeHandle, (char*)streamPartId, (char*)content, contentLength, userData); } void closeSharedLibrary() { @@ -60,3 +93,51 @@ void proxyClientDisconnectWrapper(const ProxyResult** result, uint64_t id) { uint64_t proxyClientPublishWrapper(const ProxyResult** result, uint64_t id, const char* content, uint64_t contentLength, const char* ethereumPrivateKey) { return proxyClientPublishPtr(result, id, content, contentLength, ethereumPrivateKey); } + +uint64_t streamrNodeNewWrapper(const StreamrResult** result, const char* ownEthereumAddress, const StreamrNodeConfig* config) { + return streamrNodeNewPtr(result, ownEthereumAddress, config); +} + +void streamrNodeDeleteWrapper(const StreamrResult** result, uint64_t nodeHandle) { + streamrNodeDeletePtr(result, nodeHandle); +} + +void streamrNodeStartWrapper(const StreamrResult** result, uint64_t nodeHandle) { + streamrNodeStartPtr(result, nodeHandle); +} + +void streamrNodeStopWrapper(const StreamrResult** result, uint64_t nodeHandle) { + streamrNodeStopPtr(result, nodeHandle); +} + +void streamrNodeJoinStreamPartWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const StreamrPeer* entryPoints, uint64_t numEntryPoints) { + streamrNodeJoinStreamPartPtr(result, nodeHandle, streamPartId, entryPoints, numEntryPoints); +} + +void streamrNodeLeaveStreamPartWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId) { + streamrNodeLeaveStreamPartPtr(result, nodeHandle, streamPartId); +} + +void streamrNodePublishWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const char* content, uint64_t contentLength, const char* ethereumPrivateKey) { + streamrNodePublishPtr(result, nodeHandle, streamPartId, content, contentLength, ethereumPrivateKey); +} + +uint64_t streamrNodeSubscribeWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, uintptr_t callbackKey) { + return streamrNodeSubscribePtr(result, nodeHandle, streamPartId, shimStreamrNodeMessageCallback, (void*)callbackKey); +} + +void streamrNodeUnsubscribeWrapper(const StreamrResult** result, uint64_t nodeHandle, uint64_t subscriptionHandle) { + streamrNodeUnsubscribePtr(result, nodeHandle, subscriptionHandle); +} + +uint64_t streamrNodeGetNeighborCountWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId) { + return streamrNodeGetNeighborCountPtr(result, nodeHandle, streamPartId); +} + +void streamrNodeSetProxiesWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const StreamrPeer* proxies, uint64_t numProxies, int direction, uint64_t connectionCount) { + streamrNodeSetProxiesPtr(result, nodeHandle, streamPartId, proxies, numProxies, (StreamrProxyDirection)direction, connectionCount); +} + +const StreamrPeer* streamrErrorGetPeerWrapper(const StreamrError* error) { + return error->peer; +} diff --git a/shim.h b/shim.h index 5e742f9..38614f4 100644 --- a/shim.h +++ b/shim.h @@ -2,6 +2,7 @@ #define SHIM_H #include "streamrproxyclient.h" +#include "streamrnode.h" #ifdef __cplusplus extern "C" { @@ -19,6 +20,25 @@ uint64_t proxyClientConnectWrapper(const ProxyResult** result, uint64_t id, cons void proxyClientDisconnectWrapper(const ProxyResult** result, uint64_t id); uint64_t proxyClientPublishWrapper(const ProxyResult** result, uint64_t id, const char* content, uint64_t contentLength, const char* ethereumPrivateKey); +// cgo cannot address members of anonymous unions, so the StreamrError +// peer/proxy union member is read through this accessor. +const StreamrPeer* streamrErrorGetPeerWrapper(const StreamrError* error); + +// Full-node API (streamrnode.h). streamrNodeSubscribeWrapper installs the +// cgo-exported Go trampoline as the message callback; callbackKey is the +// Go-side registry key carried through userData. +uint64_t streamrNodeNewWrapper(const StreamrResult** result, const char* ownEthereumAddress, const StreamrNodeConfig* config); +void streamrNodeDeleteWrapper(const StreamrResult** result, uint64_t nodeHandle); +void streamrNodeStartWrapper(const StreamrResult** result, uint64_t nodeHandle); +void streamrNodeStopWrapper(const StreamrResult** result, uint64_t nodeHandle); +void streamrNodeJoinStreamPartWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const StreamrPeer* entryPoints, uint64_t numEntryPoints); +void streamrNodeLeaveStreamPartWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId); +void streamrNodePublishWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const char* content, uint64_t contentLength, const char* ethereumPrivateKey); +uint64_t streamrNodeSubscribeWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, uintptr_t callbackKey); +void streamrNodeUnsubscribeWrapper(const StreamrResult** result, uint64_t nodeHandle, uint64_t subscriptionHandle); +uint64_t streamrNodeGetNeighborCountWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId); +void streamrNodeSetProxiesWrapper(const StreamrResult** result, uint64_t nodeHandle, const char* streamPartId, const StreamrPeer* proxies, uint64_t numProxies, int direction, uint64_t connectionCount); + #ifdef __cplusplus } #endif diff --git a/streamrcommon.h b/streamrcommon.h new file mode 100644 index 0000000..c55dcf1 --- /dev/null +++ b/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/streamrnode.go b/streamrnode.go new file mode 100644 index 0000000..e744839 --- /dev/null +++ b/streamrnode.go @@ -0,0 +1,248 @@ +package streamrproxyclient + +// #include +// #include "shim.h" +import "C" +import ( + "sync" + "unsafe" +) + +// Error codes of the full-node API (streamrnode.h). +const ( + 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" +) + +// Proxy direction values of StreamrNode.SetProxies; match +// StreamrProxyDirection in streamrnode.h. +const ( + PROXY_DIRECTION_PUBLISH = 0 + PROXY_DIRECTION_SUBSCRIBE = 1 +) + +// MessageCallback is invoked for every message received on a subscribed +// stream part. It runs on an internal network thread: return quickly and +// do not call StreamrNode methods from inside it. content is a copy the +// callback owns. +type MessageCallback func(streamPartId string, content []byte) + +// The registry that routes the C message callback back to the Go +// closures; keyed by the value passed through the C userData pointer. +var ( + nodeCallbacksMutex sync.RWMutex + nodeCallbacks = map[uintptr]MessageCallback{} + nextCallbackKey uintptr +) + +//export goStreamrNodeMessage +func goStreamrNodeMessage(nodeHandle C.uint64_t, streamPartId *C.char, content *C.char, contentLength C.uint64_t, userData unsafe.Pointer) { + nodeCallbacksMutex.RLock() + callback := nodeCallbacks[uintptr(userData)] + nodeCallbacksMutex.RUnlock() + if callback != nil { + callback(C.GoString(streamPartId), C.GoBytes(unsafe.Pointer(content), C.int(contentLength))) + } +} + +// StreamrNodeConfig configures a StreamrNode. The zero value means: no +// entry points (the node starts a new network and acts as its first +// entry point), no websocket server (client-only operation — the mode +// intended for mobile devices). +type StreamrNodeConfig struct { + EntryPoints []Proxy + WebsocketPort uint16 + WebsocketHost string // empty = default (127.0.0.1) + AcceptProxyConnections bool +} + +// StreamrNode is the full network node (streamrnode.h). +type StreamrNode struct { + nodeHandle C.uint64_t + callbackKeys []uintptr +} + +func firstError(result *C.StreamrResult) *ProxyClientError { + if result != nil && result.numErrors > 0 { + firstError := (*C.Error)(unsafe.Pointer(result.errors)) + err := NewProxyClientError(firstError) + C.proxyClientResultDeleteWrapper(result) + return err + } + C.proxyClientResultDeleteWrapper(result) + return nil +} + +// makeCPeers builds a C-allocated StreamrPeer array (cgo forbids Go +// pointers inside structs passed to C). freeCPeers releases it after the +// call — the C API copies what it needs before returning. +func makeCPeers(peers []Proxy) *C.StreamrPeer { + if len(peers) == 0 { + return nil + } + arr := (*C.StreamrPeer)(C.malloc(C.size_t(len(peers)) * C.size_t(unsafe.Sizeof(C.StreamrPeer{})))) + slice := unsafe.Slice(arr, len(peers)) + for i, peer := range peers { + slice[i] = C.StreamrPeer{ + websocketUrl: C.CString(peer.WebsocketUrl), + ethereumAddress: C.CString(peer.EthereumAddress), + } + } + return arr +} + +func freeCPeers(arr *C.StreamrPeer, count int) { + if arr == nil { + return + } + slice := unsafe.Slice(arr, count) + for i := range slice { + C.free(unsafe.Pointer(slice[i].websocketUrl)) + C.free(unsafe.Pointer(slice[i].ethereumAddress)) + } + C.free(unsafe.Pointer(arr)) +} + +// NewStreamrNode creates a network node (not yet started). config may be +// nil, which is equivalent to the zero config. +func NewStreamrNode(ownEthereumAddress string, config *StreamrNodeConfig) (*StreamrNode, *ProxyClientError) { + if config == nil { + config = &StreamrNodeConfig{} + } + cEntryPoints := makeCPeers(config.EntryPoints) + defer freeCPeers(cEntryPoints, len(config.EntryPoints)) + cConfig := C.StreamrNodeConfig{ + entryPoints: cEntryPoints, + numEntryPoints: C.uint64_t(len(config.EntryPoints)), + websocketPort: C.uint16_t(config.WebsocketPort), + acceptProxyConnections: C.bool(config.AcceptProxyConnections), + } + if config.WebsocketHost != "" { + cConfig.websocketHost = C.CString(config.WebsocketHost) + } + var result *C.StreamrResult + nodeHandle := C.streamrNodeNewWrapper(&result, C.CString(ownEthereumAddress), &cConfig) + if err := firstError(result); err != nil { + return nil, err + } + return &StreamrNode{nodeHandle: nodeHandle}, nil +} + +// Close stops the node if it is still running and releases it, together +// with the message-callback registrations it owns. Callbacks stay +// registered until here (not merely until Unsubscribe): the C API may +// still dispatch a message whose delivery already started when +// Unsubscribe returns. +func (n *StreamrNode) Close() *ProxyClientError { + var result *C.StreamrResult + C.streamrNodeDeleteWrapper(&result, n.nodeHandle) + err := firstError(result) + nodeCallbacksMutex.Lock() + for _, key := range n.callbackKeys { + delete(nodeCallbacks, key) + } + nodeCallbacksMutex.Unlock() + n.callbackKeys = nil + return err +} + +// Start starts the node and joins the layer-0 network; blocks until +// connected. A node cannot be started twice or restarted after Stop. +func (n *StreamrNode) Start() *ProxyClientError { + var result *C.StreamrResult + C.streamrNodeStartWrapper(&result, n.nodeHandle) + return firstError(result) +} + +// Stop stops the node and leaves the network. Terminal; idempotent. +func (n *StreamrNode) Stop() *ProxyClientError { + var result *C.StreamrResult + C.streamrNodeStopWrapper(&result, n.nodeHandle) + return firstError(result) +} + +// JoinStreamPart joins a stream part, optionally through known stream +// part entry points. +func (n *StreamrNode) JoinStreamPart(streamPartId string, entryPoints []Proxy) *ProxyClientError { + cEntryPoints := makeCPeers(entryPoints) + defer freeCPeers(cEntryPoints, len(entryPoints)) + var result *C.StreamrResult + C.streamrNodeJoinStreamPartWrapper(&result, n.nodeHandle, C.CString(streamPartId), cEntryPoints, C.uint64_t(len(entryPoints))) + return firstError(result) +} + +// LeaveStreamPart leaves a stream part. +func (n *StreamrNode) LeaveStreamPart(streamPartId string) *ProxyClientError { + var result *C.StreamrResult + C.streamrNodeLeaveStreamPartWrapper(&result, n.nodeHandle, C.CString(streamPartId)) + return firstError(result) +} + +// Publish publishes a message to a stream part (joining it first if +// needed). Pass an empty ethereumPrivateKey for an unsigned message. +func (n *StreamrNode) Publish(streamPartId string, data []byte, ethereumPrivateKey string) *ProxyClientError { + var key *C.char + if ethereumPrivateKey != "" { + key = C.CString(ethereumPrivateKey) + } + var result *C.StreamrResult + C.streamrNodePublishWrapper(&result, n.nodeHandle, C.CString(streamPartId), (*C.char)(unsafe.Pointer(&data[0])), C.uint64_t(len(data)), key) + return firstError(result) +} + +// Subscribe subscribes to a stream part (joining it first if needed); +// callback is invoked for every message received on it. Returns a +// subscription handle for Unsubscribe. +func (n *StreamrNode) Subscribe(streamPartId string, callback MessageCallback) (uint64, *ProxyClientError) { + nodeCallbacksMutex.Lock() + nextCallbackKey++ + key := nextCallbackKey + nodeCallbacks[key] = callback + nodeCallbacksMutex.Unlock() + var result *C.StreamrResult + subscriptionHandle := C.streamrNodeSubscribeWrapper(&result, n.nodeHandle, C.CString(streamPartId), C.uintptr_t(key)) + if err := firstError(result); err != nil { + nodeCallbacksMutex.Lock() + delete(nodeCallbacks, key) + nodeCallbacksMutex.Unlock() + return 0, err + } + n.callbackKeys = append(n.callbackKeys, key) + return uint64(subscriptionHandle), nil +} + +// Unsubscribe removes a subscription. The callback may still be invoked +// for messages whose dispatch already started when this returns; its +// registration is released at Close. +func (n *StreamrNode) Unsubscribe(subscriptionHandle uint64) *ProxyClientError { + var result *C.StreamrResult + C.streamrNodeUnsubscribeWrapper(&result, n.nodeHandle, C.uint64_t(subscriptionHandle)) + return firstError(result) +} + +// NeighborCount returns the number of neighbors the node currently has +// in a stream part's delivery topology. +func (n *StreamrNode) NeighborCount(streamPartId string) (uint64, *ProxyClientError) { + var result *C.StreamrResult + count := C.streamrNodeGetNeighborCountWrapper(&result, n.nodeHandle, C.CString(streamPartId)) + if err := firstError(result); err != nil { + return 0, err + } + return uint64(count), nil +} + +// SetProxies connects the stream part through proxy nodes instead of +// joining its topology; an empty proxy list disconnects it. +// connectionCount 0 means "connect to all given proxies". +func (n *StreamrNode) SetProxies(streamPartId string, proxies []Proxy, direction int, connectionCount uint64) *ProxyClientError { + cProxies := makeCPeers(proxies) + defer freeCPeers(cProxies, len(proxies)) + var result *C.StreamrResult + C.streamrNodeSetProxiesWrapper(&result, n.nodeHandle, C.CString(streamPartId), cProxies, C.uint64_t(len(proxies)), C.int(direction), C.uint64_t(connectionCount)) + return firstError(result) +} diff --git a/streamrnode.h b/streamrnode.h new file mode 100644 index 0000000..cd5dfeb --- /dev/null +++ b/streamrnode.h @@ -0,0 +1,180 @@ +#ifndef STREAMR_NODE_H +#define STREAMR_NODE_H + +// Public C API of the full Streamr network node (milestone D of +// trackerless-network-completion-plan.md). +// +// This API lives in the same shared library as the proxy-client API +// (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 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 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 "streamrcommon.h" + +#define ERROR_NODE_NOT_FOUND "NODE_NOT_FOUND" +#define ERROR_NODE_NOT_STARTED "NODE_NOT_STARTED" +#define ERROR_NODE_ALREADY_STARTED "NODE_ALREADY_STARTED" +#define ERROR_NODE_STOPPED "NODE_STOPPED" +#define ERROR_INVALID_ENTRY_POINT_URL "INVALID_ENTRY_POINT_URL" +#define ERROR_SUBSCRIPTION_NOT_FOUND "SUBSCRIPTION_NOT_FOUND" +#define ERROR_NODE_OPERATION_FAILED "NODE_OPERATION_FAILED" + +// 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 StreamrPeer StreamrEntryPoint; + +typedef struct StreamrNodeConfig { + // Entry points of the layer-0 DHT. May be NULL when numEntryPoints + // is 0; the node then joins its own DHT and acts as the first entry + // point of a new network (it should have a websocket server port in + // that case so that others can join through it). + const StreamrEntryPoint* entryPoints; + uint64_t numEntryPoints; + // Websocket server port. 0 = do not run a websocket server; the + // node then connects out as a client (websocket client + WebRTC + // connections), which is the mode intended for mobile devices. + uint16_t websocketPort; + // Host name or IP address the websocket server advertises in the + // node's peer descriptor. NULL = "127.0.0.1". Ignored when + // websocketPort is 0. + const char* websocketHost; + // Accept incoming proxy connections from proxy clients (a node with + // this enabled cannot itself use streamrNodeSetProxies; the two + // roles are mutually exclusive, as in the TypeScript + // implementation). + bool acceptProxyConnections; +} StreamrNodeConfig; + +// Direction of a proxied stream part connection; the values match +// ProxyDirection in protos/NetworkRpc.proto. +typedef enum StreamrProxyDirection { + STREAMR_PROXY_DIRECTION_PUBLISH = 0, + STREAMR_PROXY_DIRECTION_SUBSCRIBE = 1 +} StreamrProxyDirection; + +// Message callback of streamrNodeSubscribe(). Invoked on an internal +// network thread: return quickly, and do NOT call streamrNode* / +// proxyClient* functions from inside the callback. content points to a +// buffer of contentLength bytes that is only valid during the call. +// streamPartId is in canonical form (partition without leading zeros, +// e.g. "0xabc...#1" even if the subscription said "#01"). +typedef void (*StreamrNodeMessageCallback)( + uint64_t nodeHandle, + const char* streamPartId, + const char* content, + uint64_t contentLength, + void* userData); + +// Creates a network node (not yet started; no sockets are opened). +// ownEthereumAddress is the node's identity: its node id is derived +// from it, and it is used as the publisher id of published messages. +// 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 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 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 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 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 +// points); when none are given the stream part's entry points are +// 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 StreamrResult** result, + uint64_t nodeHandle, + const char* streamPartId, + const StreamrEntryPoint* streamPartEntryPoints, + uint64_t numStreamPartEntryPoints); + +// Leaves a stream part. +EXTERN_C SHARED_EXPORT void streamrNodeLeaveStreamPart( + 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 +// may be NULL for an unsigned message; when given, the message is +// signed with ECDSA (secp256k1, EVM style) exactly like +// proxyClientPublish. Fire-and-forget: delivery into the overlay is +// not acknowledged. +EXTERN_C SHARED_EXPORT void streamrNodePublish( + const StreamrResult** result, + uint64_t nodeHandle, + const char* streamPartId, + const char* content, + uint64_t contentLength, + const char* ethereumPrivateKey); + +// Subscribes to a stream part (joining it first if needed): callback is +// 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 StreamrResult** result, + uint64_t nodeHandle, + const char* streamPartId, + StreamrNodeMessageCallback callback, + void* userData); + +// 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 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 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 +// node handle). direction selects publishing or subscribing through +// 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 StreamrResult** result, + uint64_t nodeHandle, + const char* streamPartId, + const StreamrPeer* proxies, + uint64_t numProxies, + StreamrProxyDirection direction, + uint64_t connectionCount); + +#endif diff --git a/streamrnode_test.go b/streamrnode_test.go new file mode 100644 index 0000000..284e81a --- /dev/null +++ b/streamrnode_test.go @@ -0,0 +1,157 @@ +package streamrproxyclient + +// Round-trip test of the Go StreamrNode wrapper: mirrors the C API's +// StreamrNodeTest.TwoNodesExchangeMessages — two wrapper-created nodes +// form a stream-part topology and exchange a signed and an unsigned +// message. + +import ( + "bytes" + "fmt" + "sync" + "testing" + "time" +) + +const ( + nodeEthereumAddressA = "0x1234567890123456789012345678901234567890" + nodeEthereumAddressB = "0x1234567890123456789012345678901234567892" + nodePrivateKey = "1111111111111111111111111111111111111111111111111111111111111111" + nodeStreamPartId = "0xa000000000000000000000000000000000000000#01" + nodeEntryPointPort = 44481 + topologyTimeout = 60 * time.Second + messageTimeout = 30 * time.Second + pollInterval = 200 * time.Millisecond +) + +func waitUntil(condition func() bool, timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if condition() { + return true + } + time.Sleep(pollInterval) + } + return condition() +} + +func TestStreamrNodeInvalidEthereumAddress(t *testing.T) { + lib := NewLibStreamrProxyClient() + defer lib.Close() + + _, err := NewStreamrNode("not-an-address", nil) + if err == nil { + t.Fatal("expected an error") + } + if err.Code != ERROR_INVALID_ETHEREUM_ADDRESS { + t.Fatalf("expected INVALID_ETHEREUM_ADDRESS, got %s", err.Code) + } +} + +func TestStreamrNodeTwoNodesExchangeMessages(t *testing.T) { + lib := NewLibStreamrProxyClient() + defer lib.Close() + + // Node A runs the websocket server of a new network; node B joins + // through it. + nodeA, err := NewStreamrNode(nodeEthereumAddressA, &StreamrNodeConfig{ + WebsocketPort: nodeEntryPointPort, + }) + if err != nil { + t.Fatal(err) + } + defer nodeA.Close() + if err := nodeA.Start(); err != nil { + t.Fatal(err) + } + + entryPoint := Proxy{ + WebsocketUrl: fmt.Sprintf("ws://127.0.0.1:%d", nodeEntryPointPort), + EthereumAddress: nodeEthereumAddressA, + } + nodeB, err := NewStreamrNode(nodeEthereumAddressB, &StreamrNodeConfig{ + EntryPoints: []Proxy{entryPoint}, + }) + if err != nil { + t.Fatal(err) + } + defer nodeB.Close() + if err := nodeB.Start(); err != nil { + t.Fatal(err) + } + + var mutex sync.Mutex + var receivedA, receivedB [][]byte + subA, err := nodeA.Subscribe(nodeStreamPartId, func(_ string, content []byte) { + mutex.Lock() + receivedA = append(receivedA, content) + mutex.Unlock() + }) + if err != nil { + t.Fatal(err) + } + subB, err := nodeB.Subscribe(nodeStreamPartId, func(_ string, content []byte) { + mutex.Lock() + receivedB = append(receivedB, content) + mutex.Unlock() + }) + if err != nil { + t.Fatal(err) + } + + if !waitUntil(func() bool { + countA, errA := nodeA.NeighborCount(nodeStreamPartId) + countB, errB := nodeB.NeighborCount(nodeStreamPartId) + return errA == nil && errB == nil && countA >= 1 && countB >= 1 + }, topologyTimeout) { + t.Fatal("stream part topology did not form") + } + + contains := func(received *[][]byte, message []byte) func() bool { + return func() bool { + mutex.Lock() + defer mutex.Unlock() + for _, m := range *received { + if bytes.Equal(m, message) { + return true + } + } + return false + } + } + + messageFromA := []byte("hello from A (signed)") + if err := nodeA.Publish(nodeStreamPartId, messageFromA, nodePrivateKey); err != nil { + t.Fatal(err) + } + if !waitUntil(contains(&receivedB, messageFromA), messageTimeout) { + t.Fatal("node B did not receive the message from A") + } + + messageFromB := []byte("hello from B (unsigned)") + if err := nodeB.Publish(nodeStreamPartId, messageFromB, ""); err != nil { + t.Fatal(err) + } + if !waitUntil(contains(&receivedA, messageFromB), messageTimeout) { + t.Fatal("node A did not receive the message from B") + } + + if err := nodeB.Unsubscribe(subB); err != nil { + t.Fatal(err) + } + if err := nodeB.Unsubscribe(subB); err == nil { + t.Fatal("expected SUBSCRIPTION_NOT_FOUND for a double unsubscribe") + } else if err.Code != ERROR_SUBSCRIPTION_NOT_FOUND { + t.Fatalf("expected SUBSCRIPTION_NOT_FOUND, got %s", err.Code) + } + if err := nodeA.Unsubscribe(subA); err != nil { + t.Fatal(err) + } + + if err := nodeB.Stop(); err != nil { + t.Fatal(err) + } + if err := nodeA.Stop(); err != nil { + t.Fatal(err) + } +} diff --git a/streamrproxyclient.go b/streamrproxyclient.go index 264ec5b..de821fe 100644 --- a/streamrproxyclient.go +++ b/streamrproxyclient.go @@ -6,20 +6,29 @@ package streamrproxyclient import "C" import ( "fmt" + "sync" "unsafe" ) -func openLibrary() (error) { - fileName, err := SaveLibToTempFile() - if err != nil { - return err - } - C.loadSharedLibrary(C.CString(fileName)) - return nil -} - -func closeLibrary() { - C.closeSharedLibrary() +// The shared library is loaded ONCE per process and never dlclosed: +// a node that has run leaves service threads that pin the library in +// memory, so a dlclose + fresh dlopen would load a SECOND copy and +// abort on duplicate gflags registration (folly). In-process cleanup +// and re-initialization go through streamrCleanupLibrary()/ +// streamrInitLibrary() on the one loaded copy instead. +var openLibraryOnce sync.Once + +func openLibrary() error { + var err error + openLibraryOnce.Do(func() { + var fileName string + fileName, err = SaveLibToTempFile() + if err != nil { + return + } + C.loadSharedLibrary(C.CString(fileName)) + }) + return err } // Error codes from the C library. @@ -80,8 +89,11 @@ func (e *ProxyClientError) Error() string { // NewProxyClientError creates a new ProxyClientError from a C Error. func NewProxyClientError(cError *C.Error) *ProxyClientError { var proxy *Proxy - if cError.proxy != nil { - proxy = NewProxy(C.GoString(cError.proxy.websocketUrl), C.GoString(cError.proxy.ethereumAddress)) + // The peer member sits in an anonymous union (peer/proxy spellings) + // that cgo cannot address directly; read it through the shim. + cPeer := C.streamrErrorGetPeerWrapper(cError) + if cPeer != nil { + proxy = NewProxy(C.GoString(cPeer.websocketUrl), C.GoString(cPeer.ethereumAddress)) } return &ProxyClientError{ Message: C.GoString(cError.message), @@ -136,10 +148,11 @@ func NewLibStreamrProxyClient() *LibStreamrProxyClient { return lib } -// Close cleans up the Streamr Proxy Client library. +// Close cleans up the Streamr Proxy Client library. The shared library +// itself stays loaded (see openLibrary); a later +// NewLibStreamrProxyClient re-initializes it in place. func (l *LibStreamrProxyClient) Close() { C.proxyClientCleanupLibraryWrapper() - closeLibrary() } // ProxyClient struct represents a client that connects to proxies. diff --git a/streamrproxyclient.h b/streamrproxyclient.h index b28c146..5191d90 100644 --- a/streamrproxyclient.h +++ b/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,59 +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. - */ +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 @@ -92,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. @@ -121,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,