feature: initial migration to tokio based network stack

Author: Dushistov

Cargo.toml

@@ -1,2 +1,6 @@
 [workspace]
-members = ["couchbase-lite-core-sys", "couchbase-lite", "chat-demo"]
+members = ["couchbase-lite-core-sys", "couchbase-lite", "chat-demo"]
+
+[patch.'crates-io']
+http = { git = "https://github.com/Dushistov/http", rev = "3ad3b825edc3a0d9048a6af5ab524f2fc51bde30" }
+tungstenite = { git = "https://github.com/Dushistov/tungstenite-rs", rev = "de80b9bced480d407665e46001fc2d7a6874d71e" }

chat-demo/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-couchbase-lite = { path = "../couchbase-lite" }
+couchbase-lite = { path = "../couchbase-lite", features = ["replication"] }
 serde = { version = "1.0.104", features = ["derive"] }
 tokio = { version = "0.2.6", features = ["io-std", "io-util"] }
 log = "0.4.8"

chat-demo/src/main.rs

@@ -1,6 +1,6 @@
 use couchbase_lite::{
-    fallible_streaming_iterator::FallibleStreamingIterator, use_c4_civet_web_socket_factory,
-    Database, DatabaseConfig, Document, ReplicatorState,
+    fallible_streaming_iterator::FallibleStreamingIterator, use_web_sockets, Database,
+    DatabaseConfig, Document, ReplicatorState,
 };
 use log::{error, trace};
 use serde::{Deserialize, Serialize};
@@ -24,7 +24,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .unwrap_or_else(|| "ws://192.168.1.132:4984/demo/".to_string());
     let token: Option<String> = env::args().nth(3);
 
-    use_c4_civet_web_socket_factory();
+    use_web_sockets(runtime.handle().clone());
     let (db_thread, db_exec) = run_db_thread(db_path);
     let db_exec_repl = db_exec.clone();
     db_exec.spawn(move |db| {

couchbase-lite-core-sys/build.rs

@@ -9,6 +9,7 @@ use std::{
 fn main() {
     let dst = cmake::Config::new(Path::new("couchbase-lite-core"))
         .define("DISABLE_LTO_BUILD", "True")
+        .define("SANITIZE_FOR_DEBUG_ENABLED", "False")
         .build_target("LiteCore")
         .build()
         .join("build");
@@ -100,7 +101,6 @@ fn main() {
             "fleece/FLSlice.h",
             "c4Document+Fleece.h",
             "fleece/Fleece.h",
-            "Replicator/CivetWebSocket.hh",
             "couch_lite_log_retrans.hpp",
         ],
         &out_dir.join("c4_header.rs"),

couchbase-lite-core-sys/couchbase-lite-core

@@ -1,3 +1,23 @@
 mod c4_header;
 
 pub use c4_header::*;
+
+use std::os::raw::c_void;
+
+//bindgen can not handle inline functions so
+
+#[inline]
+pub unsafe fn c4db_release(db: *mut C4Database) {
+    c4base_release(db as *mut c_void)
+}
+
+#[inline]
+#[allow(non_snake_case)]
+pub unsafe fn FLSliceResult_Release(s: FLSliceResult) {
+    _FLBuf_Release(s.buf);
+}
+
+#[inline]
+pub unsafe fn c4query_release(r: *mut C4Query) {
+    c4base_release(r as *mut c_void)
+}

couchbase-lite-core-sys/src/lib.rs

@@ -4,7 +4,10 @@ version = "0.2.0"
 authors = ["Evgeniy A. Dushistov <dushistov@mail.ru>"]
 edition = "2018"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[features]
+default = ["replication", "tls"]
+replication = ["tokio-tungstenite", "http", "tokio", "futures-util",  "tokio-tls"]
+tls = ["tokio-tungstenite/tls"]
 
 [dependencies]
 couchbase-lite-core-sys = { path = "../couchbase-lite-core-sys" }
@@ -16,7 +19,12 @@ fallible-streaming-iterator = "0.1.9"
 bitflags = "1.2.1"
 once_cell = "1.2.0"
 log = "0.4.8"
+tokio-tungstenite = { version = "0.10.1", optional = true }
+http = { version = "0.2.1", optional = true }
+tokio = { version = "0.2", default-features = false, features = ["sync", "macros"], optional = true }
+futures-util = { version = "0.3.5", optional = true }
+tokio-tls = { version = "0.3.1", optional = true }
 
 [dev-dependencies]
 tempfile = "3.0"
-env_logger = "0.7"
+env_logger = "0.7"

couchbase-lite/Cargo.toml

@@ -1,7 +1,7 @@
 use crate::{
     error::{c4error_init, Error},
     ffi::{
-        c4db_encodeJSON, c4doc_bodyAsJSON, c4doc_free, c4doc_loadRevisionBody, kDocExists,
+        c4db_encodeJSON, c4doc_bodyAsJSON, c4doc_loadRevisionBody, c4doc_release, kDocExists,
         kRevDeleted, C4Document, C4DocumentFlags, C4RevisionFlags,
     },
     fl_slice::{fl_slice_to_str_unchecked, AsFlSlice, FlSliceOwner},
@@ -175,6 +175,6 @@ impl C4DocumentOwner {
 
 impl Drop for C4DocumentOwner {
     fn drop(&mut self) {
-        unsafe { c4doc_free(self.0.as_ptr()) };
+        unsafe { c4doc_release(self.0.as_ptr()) };
     }
 }

couchbase-lite/src/document.rs

@@ -96,3 +96,8 @@ pub(crate) unsafe fn fl_slice_to_str_unchecked<'a>(s: FLSlice) -> &'a str {
     let bytes: &[u8] = slice::from_raw_parts(s.buf as *const u8, s.size);
     str::from_utf8_unchecked(bytes)
 }
+
+#[inline]
+pub(crate) unsafe fn fl_slice_to_slice<'a>(s: FLSlice) -> &'a [u8] {
+    slice::from_raw_parts(s.buf as *const u8, s.size)
+}

couchbase-lite/src/fl_slice.rs

@@ -40,13 +40,21 @@
 //! }
 //! ```
 
+macro_rules! slice_without_nul {
+    ($c:ident) => {
+        &$c[0..($c.len() - 1)]
+    };
+}
+
 mod doc_enumerator;
 mod document;
 mod error;
 mod fl_slice;
 mod log_reroute;
 mod observer;
 mod query;
+#[cfg(feature = "replication")]
+mod repl_transport;
 mod replicator;
 mod transaction;
 mod value;
@@ -60,18 +68,19 @@ pub use crate::{
 };
 pub use couchbase_lite_core_sys as ffi;
 pub use fallible_streaming_iterator;
+#[cfg(feature = "replication")]
+pub use repl_transport::use_web_sockets;
 
 use crate::{
     document::C4DocumentOwner,
     error::{c4error_init, Result},
     ffi::{
-        c4db_createIndex, c4db_free, c4db_getDocumentCount, c4db_getIndexes, c4db_open, c4doc_get,
-        c4socket_registerFactory, kC4ArrayIndex, kC4DB_Create, kC4EncryptionNone, kC4FullTextIndex,
-        kC4PredictiveIndex, kC4RevisionTrees, kC4SQLiteStorageEngine, kC4ValueIndex,
-        C4CivetWebSocketFactory, C4Database, C4DatabaseConfig, C4DatabaseFlags,
-        C4DocumentVersioning, C4EncryptionAlgorithm, C4EncryptionKey, C4IndexOptions, C4String,
-        FLTrust_kFLTrusted, FLValue, FLValueType, FLValue_AsString, FLValue_FromData,
-        FLValue_GetType,
+        c4db_createIndex, c4db_getDocumentCount, c4db_getIndexes, c4db_open, c4db_release,
+        c4doc_get, kC4ArrayIndex, kC4DB_Create, kC4EncryptionNone, kC4FullTextIndex,
+        kC4PredictiveIndex, kC4RevisionTrees, kC4SQLiteStorageEngine, kC4ValueIndex, C4Database,
+        C4DatabaseConfig, C4DatabaseFlags, C4DocumentVersioning, C4EncryptionAlgorithm,
+        C4EncryptionKey, C4IndexOptions, C4String, FLTrust_kFLTrusted, FLValue, FLValueType,
+        FLValue_AsString, FLValue_FromData, FLValue_GetType,
     },
     fl_slice::{fl_slice_to_str_unchecked, AsFlSlice, FlSliceOwner},
     log_reroute::DB_LOGGER,
@@ -114,11 +123,6 @@ impl Default for DatabaseConfig {
     }
 }
 
-/// use embedded web-socket library
-pub fn use_c4_civet_web_socket_factory() {
-    unsafe { c4socket_registerFactory(C4CivetWebSocketFactory) };
-}
-
 /// A connection to a couchbase-lite database.
 pub struct Database {
     inner: DbInner,
@@ -145,7 +149,7 @@ impl Drop for Database {
             repl.stop();
         }
         self.db_observers.clear();
-        unsafe { c4db_free(self.inner.0.as_ptr()) };
+        unsafe { c4db_release(self.inner.0.as_ptr()) };
     }
 }
 
@@ -272,8 +276,8 @@ impl Database {
     where
         F: FnMut(ReplicatorState) + Send + 'static,
     {
-        self.db_replicator =
-            Some(Replicator::new(
+        let mut db_replicator =
+            Replicator::new(
                 self,
                 url,
                 token,
@@ -283,7 +287,9 @@ impl Database {
                         error!("replicator status change: invalid status {}", err);
                     }
                 },
-            )?);
+            )?;
+        db_replicator.start();
+        self.db_replicator = Some(db_replicator);
         self.replicator_params = Some(ReplicatorParams {
             url: url.into(),
             token: token.map(str::to_string),