|
|
@ -17,14 +17,15 @@ fn test_change_stream() {
|
|
|
|
let guard = thread::spawn(move || {
|
|
|
|
let guard = thread::spawn(move || {
|
|
|
|
let client = cloned_pool.pop();
|
|
|
|
let client = cloned_pool.pop();
|
|
|
|
let 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": 1_000}).unwrap();
|
|
|
|
let stream = collection.watch(&doc!{"$match": {"$gte": 10}}, &doc!{"maxAwaitTimeMS": 1_000}).unwrap();
|
|
|
|
let mut counter = 0;
|
|
|
|
|
|
|
|
|
|
|
|
let mut counter = 10;
|
|
|
|
for x in stream {
|
|
|
|
for x in stream {
|
|
|
|
let c = x.unwrap().get_document("fullDocument").unwrap().get_i32("c").unwrap();
|
|
|
|
let c = x.unwrap().get_document("fullDocument").unwrap().get_i32("c").unwrap();
|
|
|
|
if c == counter {
|
|
|
|
if c == counter {
|
|
|
|
counter += 1;
|
|
|
|
counter += 1;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if counter == 15 {
|
|
|
|
if counter == 25 {
|
|
|
|
break;
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
|
|
|
};
|
|
|
@ -33,11 +34,11 @@ fn test_change_stream() {
|
|
|
|
|
|
|
|
|
|
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
|
|
|
|
|
|
|
|
for i in 0..15 {
|
|
|
|
for i in 0..25 {
|
|
|
|
collection.insert(&doc! {"c": i}, None).unwrap();
|
|
|
|
collection.insert(&doc! {"c": i}, None).unwrap();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
assert_eq!(15, guard.join().unwrap());
|
|
|
|
assert_eq!(25, guard.join().unwrap());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|