Merge pull request #60 from tstenner/checkquery

Raise an error when attempting stream resolution with an malformed query

Author: tstenner

include/lsl_cpp.h

@@ -26,6 +26,9 @@ namespace lsl {
    #include "lsl_c.h"
 #endif
 
+/// Assert that no error happened; throw appropriate exception otherwise
+int32_t check_error(int32_t ec);
+
     /// Constant to indicate that a stream has variable sampling rate.
     const double IRREGULAR_RATE = 0.0; 
 
@@ -649,7 +652,11 @@ namespace lsl {
     * @return A vector of stream info objects (excluding their desc field), any of which can 
     *         subsequently be used to open an inlet. The full info can be retrieve from the inlet.
     */
-    inline std::vector<stream_info> resolve_streams(double wait_time=1.0) { lsl_streaminfo buffer[1024]; return std::vector<stream_info>(&buffer[0],&buffer[lsl_resolve_all(buffer,sizeof(buffer),wait_time)]); }
+	inline std::vector<stream_info> resolve_streams(double wait_time = 1.0) {
+		lsl_streaminfo buffer[1024];
+		int nres = check_error(lsl_resolve_all(buffer, sizeof(buffer), wait_time));
+		return std::vector<stream_info>(&buffer[0], &buffer[nres]);
+	}
 
     /** Resolve all streams with a specific value for a given property.
     * If the goal is to resolve a specific stream, this method is preferred over resolving all streams and then selecting the desired one.
@@ -661,7 +668,14 @@ namespace lsl {
     * @return A vector of matching stream info objects (excluding their meta-data), any of 
     *         which can subsequently be used to open an inlet.
     */
-    inline std::vector<stream_info> resolve_stream(const std::string &prop, const std::string &value, int32_t minimum=1, double timeout=FOREVER) { lsl_streaminfo buffer[1024]; return std::vector<stream_info>(&buffer[0],&buffer[lsl_resolve_byprop(buffer,sizeof(buffer),(prop.c_str()),(value.c_str()),minimum,timeout)]); }
+	inline std::vector<stream_info> resolve_stream(const std::string &prop,
+		const std::string &value, int32_t minimum = 1, double timeout = FOREVER) {
+		lsl_streaminfo buffer[1024];
+		int nres = check_error(lsl_resolve_byprop(
+			buffer, sizeof(buffer), prop.c_str(), value.c_str(), minimum, timeout));
+		return std::vector<stream_info>(
+			&buffer[0], &buffer[nres]);
+	}
 
 	/** Resolve all streams that match a given predicate.
 	 *
@@ -677,7 +691,13 @@ namespace lsl {
 	 * @return A vector of matching stream info objects (excluding their meta-data), any of
 	 *         which can subsequently be used to open an inlet.
 	 */
-	inline std::vector<stream_info> resolve_stream(const std::string &pred, int32_t minimum=1, double timeout=FOREVER) { lsl_streaminfo buffer[1024]; return std::vector<stream_info>(&buffer[0],&buffer[lsl_resolve_bypred(buffer,sizeof(buffer),(pred.c_str()),minimum,timeout)]); }
+	inline std::vector<stream_info> resolve_stream(
+		const std::string &pred, int32_t minimum = 1, double timeout = FOREVER) {
+		lsl_streaminfo buffer[1024];
+		int nres =
+			check_error(lsl_resolve_bypred(buffer, sizeof(buffer), pred.c_str(), minimum, timeout));
+		return std::vector<stream_info>(&buffer[0], &buffer[nres]);
+	}
 
 
     // ======================
@@ -687,7 +707,6 @@ namespace lsl {
     /** A stream inlet.
     * Inlets are used to receive streaming data (and meta-data) from the lab network.
     */  
-    void check_error(int32_t ec);
     class stream_inlet {
     public:
         /** Construct a new stream inlet from a resolved stream info.
@@ -1307,7 +1326,7 @@ namespace lsl {
     * Check error codes returned from the C interface 
     * and translate into appropriate exceptions.
     */
-    inline void check_error(int32_t ec) {
+    inline int32_t check_error(int32_t ec) {
         if (ec<0) {
             switch(ec) {
                 case lsl_timeout_error:
@@ -1322,6 +1341,7 @@ namespace lsl {
                     throw std::runtime_error("An unknown error has occurred.");
             }
         }
+		return ec;
     }
 
 } // end namespace

src/lsl_resolver_c.cpp

@@ -17,17 +17,7 @@ using namespace lsl;
 *					  The recommended default value is 5.0.
 */
 LIBLSL_C_API lsl_continuous_resolver lsl_create_continuous_resolver(double forget_after) {
-	try {
-		// create a new resolver
-		resolver_impl *resolver = new resolver_impl();
-		// start it with the given query
-		std::ostringstream os; os << "session_id='" << api_config::get_instance()->session_id() << "'";
-		resolver->resolve_continuous(os.str(),forget_after);
-		return resolver;
-	} catch(std::exception &e) {
-		LOG_F(ERROR, "Error while creating a continuous_resolver: %s", e.what());
-		return nullptr;
-	}
+	return resolver_impl::create_resolver(forget_after);
 }
 
 /**
@@ -40,17 +30,7 @@ LIBLSL_C_API lsl_continuous_resolver lsl_create_continuous_resolver(double forge
 *					  The recommended default value is 5.0.
 */
 LIBLSL_C_API lsl_continuous_resolver lsl_create_continuous_resolver_byprop(const char *prop, const char *value, double forget_after) {
-	try {
-		// create a new resolver
-		resolver_impl *resolver = new resolver_impl();
-		// start it with the given query
-		std::ostringstream os; os << "session_id='" << api_config::get_instance()->session_id() << "' and " << prop << "='" << value << "'";
-		resolver->resolve_continuous(os.str(),forget_after);
-		return resolver;
-	} catch(std::exception &e) {
-		LOG_F(ERROR, "Error while creating a continuous_resolver: %s", e.what());
-		return nullptr;
-	}
+	return resolver_impl::create_resolver(forget_after, prop, value);
 }
 
 /**
@@ -62,17 +42,7 @@ LIBLSL_C_API lsl_continuous_resolver lsl_create_continuous_resolver_byprop(const
 *					  The recommended default value is 5.0.
 */
 LIBLSL_C_API lsl_continuous_resolver lsl_create_continuous_resolver_bypred(const char *pred, double forget_after) {
-	try {
-		// create a new resolver
-		resolver_impl *resolver = new resolver_impl();
-		// start it with the given query
-		std::ostringstream os; os << "session_id='" << api_config::get_instance()->session_id() << "' and " << pred;
-		resolver->resolve_continuous(os.str(),forget_after);
-		return resolver;
-	} catch(std::exception &e) {
-		LOG_F(ERROR, "Error while creating a continuous_resolver: %s", e.what());
-		return nullptr;
-	}
+	return resolver_impl::create_resolver(forget_after, pred, nullptr);
 }
 
 /**
@@ -86,13 +56,10 @@ LIBLSL_C_API lsl_continuous_resolver lsl_create_continuous_resolver_bypred(const
 */
 LIBLSL_C_API int32_t lsl_resolver_results(lsl_continuous_resolver res, lsl_streaminfo *buffer, uint32_t buffer_elements) {
 	try {
-		// query it
-		resolver_impl *resolver = res;
-		std::vector<stream_info_impl> tmp = resolver->results();
+		std::vector<stream_info_impl> tmp = res->results(buffer_elements);
 		// allocate new stream_info_impl's and assign to the buffer
-		uint32_t result = buffer_elements<tmp.size() ? buffer_elements : (uint32_t)tmp.size();
-		for (uint32_t k = 0; k < result; k++) buffer[k] = new stream_info_impl(tmp[k]);
-		return result;
+		for (uint32_t k = 0; k < tmp.size(); k++) buffer[k] = new stream_info_impl(tmp[k]);
+		return tmp.size();
 	} catch(std::exception &e) {
 		LOG_F(WARNING, "Unexpected error querying lsl_resolver_results: %s", e.what());
 		return lsl_internal_error;
@@ -162,12 +129,8 @@ LIBLSL_C_API int32_t lsl_resolve_all(lsl_streaminfo *buffer, uint32_t buffer_ele
 */
 LIBLSL_C_API int32_t lsl_resolve_byprop(lsl_streaminfo *buffer, uint32_t buffer_elements, const char *prop, const char *value, int32_t minimum, double timeout) {
 	try {
-		// create a new resolver
-		resolver_impl resolver;
-		// build a new query.
-		std::ostringstream os; os << "session_id='" << api_config::get_instance()->session_id() << "' and " << prop << "='" << value << "'";
-		// invoke it
-		std::vector<stream_info_impl> tmp = resolver.resolve_oneshot(os.str(),minimum,timeout);
+		std::string query{resolver_impl::build_query(prop, value)};
+		auto tmp = resolver_impl().resolve_oneshot(query, minimum, timeout);
 		// allocate new stream_info_impl's and assign to the buffer
 		uint32_t result = buffer_elements<tmp.size() ? buffer_elements : (uint32_t)tmp.size();
 		for (uint32_t k = 0; k < result; k++) buffer[k] = new stream_info_impl(tmp[k]);
@@ -195,12 +158,8 @@ LIBLSL_C_API int32_t lsl_resolve_byprop(lsl_streaminfo *buffer, uint32_t buffer_
 */
 LIBLSL_C_API int32_t lsl_resolve_bypred(lsl_streaminfo *buffer, uint32_t buffer_elements, const char *pred, int32_t minimum, double timeout) {
 	try {
-		// create a new resolver
-		resolver_impl resolver;
-		// build a new query.
-		std::ostringstream os; os << "session_id='" << api_config::get_instance()->session_id() << "' and " << pred;
-		// invoke it
-		std::vector<stream_info_impl> tmp = resolver.resolve_oneshot(os.str(),minimum,timeout);
+		std::string query{resolver_impl::build_query(pred)};
+		auto tmp = resolver_impl().resolve_oneshot(query, minimum, timeout);
 		// allocate new stream_info_impl's and assign to the buffer
 		uint32_t result = buffer_elements<tmp.size() ? buffer_elements : (uint32_t)tmp.size();
 		for (uint32_t k = 0; k < result; k++) buffer[k] = new stream_info_impl(tmp[k]);

src/resolver_impl.cpp

@@ -63,6 +63,35 @@ resolver_impl::resolver_impl()
 	}
 }
 
+void check_query(const std::string& query) {
+	try {
+		pugi::xpath_query(query.c_str());
+	} catch (std::exception &e) {
+		throw std::invalid_argument((("Invalid query '" + query) += "': ") += e.what());
+	}
+}
+
+std::string resolver_impl::build_query(const char* pred_or_prop, const char* value)
+{
+	std::string query("session_id='");
+	query += api_config::get_instance()->session_id();
+	query += '\'';
+	if (pred_or_prop) (query += " and ") += pred_or_prop;
+	if (value) ((query += "='") += value) += '\'';
+	return query;
+}
+
+resolver_impl *resolver_impl::create_resolver(
+	double forget_after, const char *pred_or_prop, const char *value) noexcept {
+	try {
+		auto *resolver = new resolver_impl();
+		resolver->resolve_continuous(build_query(pred_or_prop, value), forget_after);
+		return resolver;
+	} catch (std::exception &e) {
+		LOG_F(ERROR, "Error while creating a continuous_resolver: %s", e.what());
+		return nullptr;
+	}
+}
 
 // === resolve functions ===
 
@@ -71,6 +100,7 @@ resolver_impl::resolver_impl()
 * Blocks until at least the minimum number of streams has been resolved, or the timeout fires, or the resolve has been cancelled.
 */
 std::vector<stream_info_impl> resolver_impl::resolve_oneshot(const std::string &query, int minimum, double timeout, double minimum_time) {
+	check_query(query);
 	// reset the IO service & set up the query parameters
 	io_->restart();
 	query_ = query;
@@ -104,6 +134,7 @@ std::vector<stream_info_impl> resolver_impl::resolve_oneshot(const std::string &
 }
 
 void resolver_impl::resolve_continuous(const std::string &query, double forget_after) {
+	check_query(query);
 	// reset the IO service & set up the query parameters
 	io_->restart();
 	query_ = query;
@@ -120,17 +151,18 @@ void resolver_impl::resolve_continuous(const std::string &query, double forget_a
 }
 
 /// Get the current set of results (e.g., during continuous operation).
-std::vector<stream_info_impl> resolver_impl::results() {
+std::vector<stream_info_impl> resolver_impl::results(uint32_t max_results) {
 	std::vector<stream_info_impl> output;
 	lslboost::lock_guard<lslboost::mutex> lock(results_mut_);
 	double expired_before = lsl_clock() - forget_after_;
-	for(result_container::iterator i=results_.begin(); i!=results_.end();) {
-		if (i->second.second < expired_before) {
-			result_container::iterator tmp = i++;
-			results_.erase(tmp);
-		} else {
-			output.push_back(i->second.first);
-			i++;
+
+	for (auto it = results_.begin(); it != results_.end();) {
+		if (it->second.second < expired_before)
+			it = results_.erase(it);
+		else {
+			if (output.size() < max_results)
+				output.push_back(it->second.first);
+			it++;
 		}
 	}
 	return output;

src/resolver_impl.h

@@ -44,6 +44,25 @@ namespace lsl {
 		*/
 		resolver_impl();
 
+		/** Build a query string
+		 *
+		 * @param pred_or_prop an entire predicate if value isn't set or the
+		 * name of the property, e.g. "foo='bar'" / "foo" (+value set as "bar")
+		 * @param value the value for the property parameter
+		 */
+		static std::string build_query(
+			const char *pred_or_prop = nullptr, const char *value = nullptr);
+
+		/** Create a resolver object with optionally a predicate or property + value
+		 *
+		 * @param pred_or_prop an entire predicate of value isn't set or the
+		 * name of the property, e.g. "foo='bar'" / "foo" (+value set as "bar")
+		 * @param value the value for the property parameter
+		 * @return A pointer to the resolver on success or nullptr on error
+		 */
+		static resolver_impl *create_resolver(double forget_after,
+			const char *pred_or_prop = nullptr, const char *value = nullptr) noexcept;
+
 		/// Destructor.
 		/// Cancels any ongoing processes and waits until they finish.
 		~resolver_impl();
@@ -75,7 +94,7 @@ namespace lsl {
 		void resolve_continuous(const std::string &query, double forget_after=5.0);
 
 		/// Get the current set of results (e.g., during continuous operation).
-		std::vector<stream_info_impl> results();
+		std::vector<stream_info_impl> results(uint32_t max_results = 4294967295);
 
 		/// Tear down any ongoing operations and render the resolver unusable.
 		/// This can be used to cancel a blocking resolve_oneshot() from another thread (e.g., to initiate teardown of the object).

testing/discovery.cpp

@@ -23,4 +23,8 @@ TEST_CASE("resolve from streaminfo", "[resolver][streaminfo][basic]") {
 	lsl::stream_inlet(outlet.info());
 }
 
+TEST_CASE("Invalid queries are caught before sending the query", "[resolver][streaminfo][basic]") {
+	REQUIRE_THROWS(lsl::resolve_stream("invalid'query", 0, 0.1));
+}
+
 } // namespace