python: make use of Filter, Where in DhtRunner.get

include/opendht/callbacks.h

@@ -99,12 +99,14 @@ using DoneCallback = std::function<void(bool success, const std::vector<std::sha
 typedef void (*DoneCallbackRaw)(bool, std::vector<std::shared_ptr<Node>>*, void *user_data);
 typedef void (*ShutdownCallbackRaw)(void *user_data);
 typedef void (*DoneCallbackSimpleRaw)(bool, void *user_data);
+typedef bool (*FilterRaw)(const Value&, void *user_data);
 
 using DoneCallbackSimple = std::function<void(bool success)>;
 
 OPENDHT_PUBLIC ShutdownCallback bindShutdownCb(ShutdownCallbackRaw shutdown_cb_raw, void* user_data);
 OPENDHT_PUBLIC DoneCallback bindDoneCb(DoneCallbackSimple donecb);
 OPENDHT_PUBLIC DoneCallback bindDoneCb(DoneCallbackRaw raw_cb, void* user_data);
 OPENDHT_PUBLIC DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw raw_cb, void* user_data);
+OPENDHT_PUBLIC Value::Filter bindFilterRaw(FilterRaw raw_filter, void* user_data);
 
 }

python/opendht.pyx

@@ -58,10 +58,12 @@ cdef inline void shutdown_callback(void* user_data) with gil:
     ref.Py_DECREF(cbs)
 
 cdef inline bool get_callback(shared_ptr[cpp.Value] value, void *user_data) with gil:
-    cb = (<object>user_data)['get']
+    cbs = <object>user_data
+    cb = cbs['get']
+    f = cbs['filter'] if 'filter' in cbs else None
     pv = Value()
     pv._value = value
-    return cb(pv)
+    return cb(pv) if not f or f(pv) else True
 
 cdef inline void done_callback(bool done, cpp.vector[shared_ptr[cpp.Node]]* nodes, void *user_data) with gil:
     node_ids = []
@@ -513,7 +515,7 @@ cdef class DhtRunner(_WithID):
             stats.append(n)
         return stats
 
-    def get(self, InfoHash key, get_cb=None, done_cb=None):
+    def get(self, InfoHash key, get_cb=None, done_cb=None, filter=None, Where where=None):
         """Retreive values associated with a key on the DHT.
 
         key     -- the key for which to search
@@ -523,9 +525,12 @@ cdef class DhtRunner(_WithID):
                    operation is completed.
         """
         if get_cb:
-            cb_obj = {'get':get_cb, 'done':done_cb}
+            cb_obj = {'get':get_cb, 'done':done_cb, 'filter':filter}
             ref.Py_INCREF(cb_obj)
-            self.thisptr.get().get(key._infohash, cpp.bindGetCb(get_callback, <void*>cb_obj), cpp.bindDoneCb(done_callback, <void*>cb_obj))
+            self.thisptr.get().get(key._infohash, cpp.bindGetCb(get_callback, <void*>cb_obj),
+                    cpp.bindDoneCb(done_callback, <void*>cb_obj),
+                    cpp.nullptr, #filter implemented in the get_callback
+                    where._where)
         else:
             lock = threading.Condition()
             pending = 0
@@ -541,7 +546,7 @@ cdef class DhtRunner(_WithID):
                     lock.notify()
             with lock:
                 pending += 1
-                self.get(key, get_cb=tmp_get, done_cb=tmp_done)
+                self.get(key, get_cb=tmp_get, done_cb=tmp_done, filter=filter, where=where)
                 while pending > 0:
                     lock.wait()
             return res

python/opendht_cpp.pxd

@@ -16,7 +16,7 @@
 # along with this program; If not, see <https://www.gnu.org/licenses/>.
 
 from libc.stdint cimport *
-from libcpp cimport bool
+from libcpp cimport bool, nullptr_t, nullptr
 from libcpp.string cimport string
 from libcpp.vector cimport vector
 from libcpp.utility cimport pair
@@ -228,7 +228,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         string getStorageLog() const
         string getRoutingTablesLog(sa_family_t af) const
         string getSearchesLog(sa_family_t af) const
-        void get(InfoHash key, GetCallback get_cb, DoneCallback done_cb)
+        void get(InfoHash key, GetCallback get_cb, DoneCallback done_cb, nullptr_t f, Where w)
         void put(InfoHash key, shared_ptr[Value] val, DoneCallback done_cb)
         ListenToken listen(InfoHash key, GetCallback get_cb)
         void cancelListen(InfoHash key, SharedListenToken token)

src/value.cpp

@@ -26,6 +26,13 @@ namespace dht {
 
 const std::string Query::QUERY_PARSE_ERROR {"Error parsing query."};
 
+Value::Filter bindFilterRaw(FilterRaw raw_filter, void* user_data) {
+    if (not raw_filter) return {};
+    return [=](const Value& value) {
+        return raw_filter(value, user_data);
+    };
+}
+
 std::ostream& operator<< (std::ostream& s, const Value& v)
 {
     s << "Value[id:" << std::hex << v.id << std::dec << " ";