refactor: deduplicate events

neqo-http3/src/server.rs

@@ -131,28 +131,21 @@ impl Http3Server {
     fn process_http3(&mut self, now: Instant) {
         qtrace!([self], "Process http3 internal.");
         let mut active_conns = self.server.active_connections();
-
-        // We need to find connections that needs to be process on http3 level.
-        let mut http3_active: Vec<ActiveConnectionRef> = self
-            .http3_handlers
-            .iter()
-            .filter_map(|(conn, handler)| {
-                if handler.borrow_mut().should_be_processed() && !active_conns.contains(conn) {
-                    Some(conn)
-                } else {
-                    None
-                }
-            })
-            .cloned()
-            .collect();
-        // For http_active connection we need to put them in neqo-transport's server
-        // waiting queue.
-        active_conns.append(&mut http3_active);
-        active_conns.dedup();
-        active_conns
-            .iter()
-            .for_each(|conn| self.server.add_to_waiting(conn));
+        active_conns.extend(
+            // We need to find connections that needs to be process on http3 level.
+            self.http3_handlers
+                .iter()
+                .filter_map(|(conn, handler)| {
+                    if handler.borrow_mut().should_be_processed() {
+                        Some(conn)
+                    } else {
+                        None
+                    }
+                })
+                .cloned(),
+        );
         for mut conn in active_conns {
+            self.server.add_to_waiting(&conn);
             self.process_events(&mut conn, now);
         }
     }

neqo-transport/src/server.rs

@@ -685,8 +685,8 @@ impl Server {
 
     /// This lists the connections that have received new events
     /// as a result of calling `process()`.
-    pub fn active_connections(&mut self) -> Vec<ActiveConnectionRef> {
-        mem::take(&mut self.active).into_iter().collect()
+    pub fn active_connections(&mut self) -> HashSet<ActiveConnectionRef> {
+        mem::take(&mut self.active)
     }
 
     pub fn add_to_waiting(&mut self, c: &ActiveConnectionRef) {

neqo-transport/tests/server.rs

@@ -309,7 +309,17 @@ fn zero_rtt() {
     // The server will have received two STREAM frames now if it processed both packets.
     let active = server.active_connections();
     assert_eq!(active.len(), 1);
-    assert_eq!(active[0].borrow().stats().frame_rx.stream, 2);
+    assert_eq!(
+        active
+            .iter()
+            .next()
+            .unwrap()
+            .borrow()
+            .stats()
+            .frame_rx
+            .stream,
+        2
+    );
 
     // Complete the handshake.  As the client was pacing 0-RTT packets, extend the time
     // a little so that the pacer doesn't prevent the Finished from being sent.
@@ -321,7 +331,17 @@ fn zero_rtt() {
     mem::drop(server.process(Some(&c4), now));
     let active = server.active_connections();
     assert_eq!(active.len(), 1);
-    assert_eq!(active[0].borrow().stats().frame_rx.stream, 2);
+    assert_eq!(
+        active
+            .iter()
+            .next()
+            .unwrap()
+            .borrow()
+            .stats()
+            .frame_rx
+            .stream,
+        2
+    );
 }
 
 #[test]