ChangeStream test done

src/change_stream.rs

@@ -60,7 +60,6 @@ impl<'a> Iterator for ChangeStream<'a> {
 
         if success == 1 {
             assert!(!bson_ptr.is_null());
-            println!("DEBUG0: {}", success);
 
             let bsonc = Bsonc::from_ptr(bson_ptr);
             match bsonc.as_document() {
@@ -68,9 +67,7 @@ impl<'a> Iterator for ChangeStream<'a> {
                 Err(error)   => return Some(Err(error.into()))
             }
         } else {
-            println!("DEBUG1: {}", success);
             let error = self.error();
-            println!("DEBUG11: {}", error);
             if error.is_empty() {
                 None
             } else {

tests/change_stream.rs

@@ -2,19 +2,42 @@
 
 use std::sync::Arc;
 use std::thread;
+use std::time::Duration;
 
 use mongo_driver::client::{ClientPool,Uri};
-use mongo_driver::Result;
 
 #[test]
 fn test_change_stream() {
     let uri      = Uri::new("mongodb://localhost:27017/").unwrap();
     let pool     = Arc::new(ClientPool::new(uri, None));
     let client   = pool.pop();
-    let mut collection = client.get_collection("rust_driver_test", "change_stream");
+    let collection = client.get_collection("rust_driver_test", "change_stream");
 
-    let stream = collection.watch(&doc!{}, &doc!{"maxAwaitTimeMS": 10_000}).unwrap();
-    let next = stream.into_iter().next().unwrap().unwrap();
-    assert_eq!(true, false);
+    let cloned_pool = pool.clone();
+    let guard = thread::spawn(move || {
+        let client     = cloned_pool.pop();
+        let collection = client.get_collection("rust_driver_test", "change_stream");
+        let stream = collection.watch(&doc!{}, &doc!{"maxAwaitTimeMS": 1_000}).unwrap();
+        let mut counter = 0;
+        for x in stream {
+            let c = x.unwrap().get_document("fullDocument").unwrap().get_i32("c").unwrap();
+            if c == counter {
+                counter += 1;
+            }
+            if counter == 15 {
+                break;
+            }
+        };
+        counter
+    });
+
+    thread::sleep(Duration::from_millis(1));
+
+    for i in 0..15 {
+        collection.insert(&doc! {"c": i}, None).unwrap();
+    }
+
+
+    assert_eq!(15, guard.join().unwrap());
 }