diff options
-rw-r--r-- | include/nostr.hpp | 5 | ||||
-rw-r--r-- | src/nostr_service.cpp | 166 | ||||
-rw-r--r-- | test/nostr_service_test.cpp | 4 |
3 files changed, 91 insertions, 84 deletions
diff --git a/include/nostr.hpp b/include/nostr.hpp index b2cdfc5..e699f21 100644 --- a/include/nostr.hpp +++ b/include/nostr.hpp @@ -177,7 +177,8 @@ public: * @brief Queries all open relay connections for events matching the given set of filters, and * returns all stored matching events returned by the relays. * @param filters The filters to use for the query. - * @returns A vector of all events matching the filters from all open relay connections. + * @returns A std::future that will eventually hold a vector of all events matching the filters + * from all open relay connections. * @remark This method runs until the relays send an EOSE message, indicating they have no more * stored events matching the given filters. When the EOSE message is received, the method * will close the subscription for each relay and return the received events. @@ -185,7 +186,7 @@ public: * set on the filters in the range 1-64, inclusive. If no valid limit is given, it will be * defaulted to 16. */ - std::vector<std::shared_ptr<Event>> queryRelays(std::shared_ptr<Filters> filters); + std::future<std::vector<std::shared_ptr<Event>>> queryRelays(std::shared_ptr<Filters> filters); /** * @brief Queries all open relay connections for events matching the given set of filters. diff --git a/src/nostr_service.cpp b/src/nostr_service.cpp index 10d9fc6..73cd602 100644 --- a/src/nostr_service.cpp +++ b/src/nostr_service.cpp @@ -182,104 +182,108 @@ tuple<vector<string>, vector<string>> NostrService::publishEvent(shared_ptr<Even return make_tuple(successfulRelays, failedRelays); }; -// TODO: Make this method return a promise. // TODO: Add a timeout to this method to prevent hanging while waiting for the relay. -vector<shared_ptr<Event>> NostrService::queryRelays(shared_ptr<Filters> filters) +future<vector<shared_ptr<Event>>> NostrService::queryRelays(shared_ptr<Filters> filters) { - if (filters->limit > 64 || filters->limit < 1) + return async(launch::async, [this, filters]() -> vector<shared_ptr<Event>> { - PLOG_WARNING << "Filters limit must be between 1 and 64, inclusive. Setting limit to 16."; - filters->limit = 16; - } + if (filters->limit > 64 || filters->limit < 1) + { + PLOG_WARNING << "Filters limit must be between 1 and 64, inclusive. Setting limit to 16."; + filters->limit = 16; + } - vector<shared_ptr<Event>> events; + vector<shared_ptr<Event>> events; - string subscriptionId = this->generateSubscriptionId(); - string request; + string subscriptionId = this->generateSubscriptionId(); + string request; - try - { - request = filters->serialize(subscriptionId); - } - catch (const invalid_argument& e) - { - PLOG_ERROR << "Failed to serialize filters - invalid object: " << e.what(); - throw e; - } - catch (const json::exception& je) - { - PLOG_ERROR << "Failed to serialize filters - JSON exception: " << je.what(); - throw je; - } + try + { + request = filters->serialize(subscriptionId); + } + catch (const invalid_argument& e) + { + PLOG_ERROR << "Failed to serialize filters - invalid object: " << e.what(); + throw e; + } + catch (const json::exception& je) + { + PLOG_ERROR << "Failed to serialize filters - JSON exception: " << je.what(); + throw je; + } - vector<future<tuple<string, bool>>> requestFutures; + vector<future<tuple<string, bool>>> requestFutures; - unordered_set<string> uniqueEventIds; + unordered_set<string> uniqueEventIds; - // Send the same query to each relay. As events trickle in from each relay, they will be added - // to the events vector. Duplicate copies of the same event will be ignored, as events are - // stored on multiple relays. The function will block until all of the relays send an EOSE or - // CLOSE message. - for (const string relay : this->_activeRelays) - { - promise<tuple<string, bool>> eosePromise; - requestFutures.push_back(move(eosePromise.get_future())); + // Send the same query to each relay. As events trickle in from each relay, they will be added + // to the events vector. Duplicate copies of the same event will be ignored, as events are + // stored on multiple relays. The function will block until all of the relays send an EOSE or + // CLOSE message. + for (const string relay : this->_activeRelays) + { + promise<tuple<string, bool>> eosePromise; + requestFutures.push_back(move(eosePromise.get_future())); - auto [uri, success] = this->_client->send( - request, - relay, - [this, &relay, &events, &eosePromise, &uniqueEventIds](string payload) - { - this->onSubscriptionMessage( - payload, - [&events, &uniqueEventIds](const string&, shared_ptr<Event> event) - { - // Check if the event is unique before adding. - if (uniqueEventIds.insert(event->id).second) + auto [uri, success] = this->_client->send( + request, + relay, + [this, &relay, &events, &eosePromise, &uniqueEventIds](string payload) + { + this->onSubscriptionMessage( + payload, + [&events, &uniqueEventIds](const string&, shared_ptr<Event> event) { - events.push_back(event); - } - }, - [relay, &eosePromise](const string&) - { - eosePromise.set_value(make_tuple(relay, true)); - }, - [relay, &eosePromise](const string&, const string&) - { - eosePromise.set_value(make_tuple(relay, false)); - }); - }); + // Check if the event is unique before adding. + if (uniqueEventIds.insert(event->id).second) + { + events.push_back(event); + } + }, + [relay, &eosePromise](const string&) + { + eosePromise.set_value(make_tuple(relay, true)); + }, + [relay, &eosePromise](const string&, const string&) + { + eosePromise.set_value(make_tuple(relay, false)); + }); + }); - if (success) - { - PLOG_INFO << "Sent query to relay " << relay; - lock_guard<mutex> lock(this->_propertyMutex); - this->_subscriptions[subscriptionId].push_back(relay); - } - else - { - PLOG_WARNING << "Failed to send query to relay " << relay; - eosePromise.set_value(make_tuple(uri, false)); + if (success) + { + PLOG_INFO << "Sent query to relay " << relay; + lock_guard<mutex> lock(this->_propertyMutex); + this->_subscriptions[subscriptionId].push_back(relay); + } + else + { + PLOG_WARNING << "Failed to send query to relay " << relay; + eosePromise.set_value(make_tuple(uri, false)); + } } - } - // Close open subscriptions and disconnect from relays after events are received. - for (auto& publishFuture : requestFutures) - { - auto [relay, isEose] = publishFuture.get(); - if (isEose) - { - PLOG_INFO << "Received EOSE message from relay " << relay; - } - else + + // Close open subscriptions and disconnect from relays after events are received. + + for (auto& publishFuture : requestFutures) { - PLOG_WARNING << "Received CLOSE message from relay " << relay; - this->closeRelayConnections({ relay }); + auto [relay, isEose] = publishFuture.get(); + if (isEose) + { + PLOG_INFO << "Received EOSE message from relay " << relay; + } + else + { + PLOG_WARNING << "Received CLOSE message from relay " << relay; + this->closeRelayConnections({ relay }); + } } - } - this->closeSubscription(subscriptionId); + this->closeSubscription(subscriptionId); - return events; + return events; + }); }; string NostrService::queryRelays( diff --git a/test/nostr_service_test.cpp b/test/nostr_service_test.cpp index 4e29960..d29fc87 100644 --- a/test/nostr_service_test.cpp +++ b/test/nostr_service_test.cpp @@ -771,7 +771,9 @@ TEST_F(NostrServiceTest, QueryRelays_ReturnsEvents_UpToEOSE) })); auto filters = make_shared<nostr::Filters>(getKind0And1TestFilters()); - auto results = nostrService->queryRelays(filters); + + // await the future and get the results + auto results = nostrService->queryRelays(filters).get(); // Check results size to ensure there are no duplicates. ASSERT_EQ(results.size(), testEvents.size()); |