From dcebded8a3cb5428ec9d4416e5c56f4f6c3f7034 Mon Sep 17 00:00:00 2001
From: JMARyA <jmarya@hydrar.de>
Date: Tue, 1 Apr 2025 13:28:04 +0200
Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=91=EF=B8=8F=20rally=20fn=20fix?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/lib.rs | 27 +++++++++++++++++++--------
 1 file changed, 19 insertions(+), 8 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index 12ee43f..75d91aa 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -29,8 +29,10 @@ pub static UNION: Lazy<
 /// This executes a thread for every item executing `f(&item) -> X`. Whatever function returns first is returned while every other thread is killed.
 pub fn rally<T: Send + Sync + 'static, F, X: Send + 'static>(items: Vec<T>, f: F) -> (T, X)
 where
-    F: Fn(&T) -> X + Send + Sync + Copy + 'static,
+    F: Fn(&T) -> Option<X> + Send + Sync + Copy + 'static,
 {
+    let item_len = items.len();
+
     let (tx, rx) = mpsc::channel();
     let items_len = items.len();
     let mut handles = Vec::new();
@@ -51,16 +53,25 @@ where
 
     drop(tx);
 
-    let (fastest_item, fastest_result, elapsed) = rx.recv().unwrap();
+    let mut count = 0;
 
-    for handle in handles {
-        // todo : threads do not get killed here
-        handle.thread().unpark();
+    while count < item_len {
+        let (fastest_item, fastest_result, elapsed) = rx.recv().unwrap();
+        count += 1;
+
+        if fastest_result.is_some() {
+            for handle in handles {
+                // todo : threads do not get killed here
+                handle.thread().unpark();
+            }
+
+            log::info!("Rally ended with {items_len} items in {elapsed:?}");
+
+            return (fastest_item, fastest_result.unwrap());
+        }
     }
 
-    log::info!("Rally ended with {items_len} items in {elapsed:?}");
-
-    (fastest_item, fastest_result)
+    panic!("No useable results in rally")
 }
 
 pub fn retry<O, F: Fn() -> Option<O>>(f: F) -> O {