infra: test to reproduce multi restart problem

see #54

couchbase-lite/tests/smoke_tests.rs → couchbase-lite/tests/functional_tests.rs

@@ -1,10 +1,11 @@
 use couchbase_lite::{
-    fallible_streaming_iterator::FallibleStreamingIterator, Database, DatabaseConfig,
-    DocEnumeratorFlags, Document, IndexType,
+    fallible_streaming_iterator::FallibleStreamingIterator, use_web_sockets, Database,
+    DatabaseConfig, DocEnumeratorFlags, Document, IndexType, ReplicatorState,
 };
 use serde::{Deserialize, Serialize};
-use std::time::SystemTime;
+use std::{thread, time::SystemTime};
 use tempfile::tempdir;
+use tokio::runtime;
 
 #[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
 #[serde(tag = "type")]
@@ -589,3 +590,61 @@ fn test_n1ql_query_with_parameter() {
     }
     tmp_dir.close().expect("Can not close tmp_dir");
 }
+
+// to reproduce bug you need server, so ignored by default
+// details https://github.com/Dushistov/couchbase-lite-rust/issues/54
+#[test]
+#[ignore]
+fn test_double_replicator_restart() {
+    let _ = env_logger::try_init();
+
+    let runtime = runtime::Builder::new_current_thread()
+        .enable_io()
+        .build()
+        .unwrap();
+    use_web_sockets(runtime.handle().clone());
+
+    let tmp_dir = tempdir().expect("Can not create tmp directory");
+    println!("we create tempdir at {}", tmp_dir.path().display());
+    let db_path = tmp_dir.path().join("a.cblite2");
+    let mut db = Database::open(&db_path, DatabaseConfig::default()).unwrap();
+
+    let (sync_tx, sync_rx) = std::sync::mpsc::channel::<()>();
+    let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(1);
+
+    {
+        let sync_tx = sync_tx.clone();
+        let handle = runtime.handle().clone();
+        db.start_replicator("ws://127.0.0.1:4984/demo/", None, move |repl_state| {
+            println!("repl_state changed: {:?}", repl_state);
+            if let ReplicatorState::Idle = repl_state {
+                sync_tx.send(()).unwrap();
+                let tx = tx.clone();
+                handle.spawn(async move {
+                    tx.send(()).await.unwrap();
+                });
+            }
+        })
+        .unwrap();
+    }
+
+    let thread_join_handle = {
+        thread::spawn(move || {
+            runtime.block_on(async {
+                rx.recv().await.unwrap();
+                println!("got async event that replicator was idle");
+                rx.recv().await.unwrap();
+            });
+        })
+    };
+    sync_rx.recv().unwrap();
+    println!("got SYNC event that replicator was idle");
+    for _ in 0..10 {
+        db.restart_replicator().unwrap();
+    }
+    println!("multi restart done");
+
+    thread_join_handle.join().unwrap();
+
+    println!("tokio done");
+}