diff --git a/src/NostrPool.cpp b/src/NostrPool.cpp index bbae4b9..0541793 100644 --- a/src/NostrPool.cpp +++ b/src/NostrPool.cpp @@ -192,6 +192,10 @@ NostrString NostrPool::subscribeMany(std::initializer_list urls, Js this->subscriptions[subId].closeCallback = closeCallback; this->subscriptions[subId].eoseCallback = eoseCallback; this->subscriptions[subId].eventCallback = eventCallback; + this->subscriptions[subId].request = json; + for (auto url : urls) { + this->subscriptions[subId].relayUrls.push_back(url); + } } for (auto url : urls) { @@ -222,6 +226,15 @@ void NostrPool::closeSubscription(NostrString subId) { this->subscriptions.erase(subId); } +NostrSubscription *NostrPool::getSubscription(NostrString subId) { + auto it = this->subscriptions.find(subId); + return it == this->subscriptions.end() ? nullptr : &it->second; +} + +bool NostrPool::hasSubscription(NostrString subId) { + return this->subscriptions.find(subId) != this->subscriptions.end(); +} + NostrRelay *NostrPool::ensureRelay(NostrString url) { NostrRelay *relay = NULL; for (auto &r : this->relays) { @@ -236,6 +249,19 @@ NostrRelay *NostrPool::ensureRelay(NostrString url) { relay = new NostrRelay(conn, url); this->relays.push_back(relay); relay->conn->addMessageListener([this, relay](NostrString message) { this->onEvent(relay, message); }); + relay->conn->addConnectionStatusListener([this, relay](ConnectionStatus status) { + if (status != ConnectionStatus::CONNECTED) return; + for (auto &entry : this->subscriptions) { + NostrSubscription &subscription = entry.second; + for (const auto &subscriptionUrl : subscription.relayUrls) { + if (NostrString_equals(subscriptionUrl, relay->url)) { + subscription.eose = false; + relay->send(subscription.request); + break; + } + } + } + }); } return relay; } diff --git a/src/NostrPool.h b/src/NostrPool.h index c164ec1..870ca85 100644 --- a/src/NostrPool.h +++ b/src/NostrPool.h @@ -32,6 +32,8 @@ namespace nostr { NostrEOSECallback eoseCallback = nullptr; NostrEventCallback eventCallback = nullptr; NostrEventStatusCallback statusCallback = nullptr; + NostrString request; + std::vector relayUrls; }; typedef struct s_EventStatusCallbackEntry { @@ -133,6 +135,16 @@ namespace nostr { */ void closeSubscription(NostrString subId); + /** + * Get a subscription by ID, or nullptr if it does not exist + */ + NostrSubscription *getSubscription(NostrString subId); + + /** + * Check whether a subscription with the given ID exists + */ + bool hasSubscription(NostrString subId); + /** * Ensure the connection to the specified relay is open * @param url The relay URL diff --git a/tests/unit/test_pool_transport.cpp b/tests/unit/test_pool_transport.cpp index 3a3d5d5..8c65446 100644 --- a/tests/unit/test_pool_transport.cpp +++ b/tests/unit/test_pool_transport.cpp @@ -16,6 +16,7 @@ class MockConnection : public nostr::Connection { std::vector sent; std::vector incoming; std::vector> messageListeners; + std::vector> statusListeners; void addMessageListener(std::function listener) override { messageListeners.push_back(listener); } void send(String message) override { sent.push_back(message); } @@ -28,8 +29,13 @@ class MockConnection : public nostr::Connection { } } bool isReady() override { return ready; } - void addConnectionStatusListener(std::function) override {} + void addConnectionStatusListener(std::function listener) override { + statusListeners.push_back(listener); + } void queue(String message) { incoming.push_back(message); } + void status(nostr::ConnectionStatus value) { + for (const auto &listener : statusListeners) listener(value); + } }; class MockTransport : public nostr::Transport { @@ -82,10 +88,16 @@ bool testPoolQueueAndTransport() { nullptr, [&](const String &id) { eoseCalled = id == subId; }); CHECK(subId.length() == 64); + CHECK(pool.hasSubscription(subId)); + CHECK(pool.getSubscription(subId) != nullptr); CHECK(transport.lastConnection != nullptr); MockConnection *connection = transport.lastConnection; pool.loop(); CHECK(!connection->sent.empty() && connection->sent.back().indexOf("\"REQ\"") >= 0); + const size_t sentBeforeReconnect = connection->sent.size(); + connection->status(nostr::ConnectionStatus::DISCONNECTED); + connection->status(nostr::ConnectionStatus::CONNECTED); + CHECK(connection->sent.size() == sentBeforeReconnect + 1); nostr::UnsignedNostrEvent unsignedEvent(1, "pool event", 1700000000); nostr::SignedNostrEvent event = unsignedEvent.sign( @@ -120,6 +132,8 @@ bool testPoolQueueAndTransport() { pool.closeSubscription(subId); pool.loop(); CHECK(connection->sent.back().indexOf("\"CLOSE\"") >= 0); + CHECK(!pool.hasSubscription(subId)); + CHECK(pool.getSubscription(subId) == nullptr); CHECK(pool.getRelays().size() == 1 && pool.getConnectedRelays()->size() == 1); pool.disconnectRelay("ws://relay.example"); CHECK(pool.getRelays().empty() && transport.connections.empty());