Move to nightly rust to access async closures

Motivation
----------
A primary design pattern that we want to use is currently challenging
due to asybc closures not being supported in stable rust. In order
to make this pattern possible we should switch to nightly and enable
the feature. async_closures should be moved to stable before we need
to release the SDK.

Changes
--------
Specify the nightly toolchain.
Enable async_closures feature.
Use async closures.

Author: chvck

.github/workflows/build.yml

@@ -29,6 +29,10 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
+      - name: Update Rust Toolchain Target
+        run: |
+          echo "targets = ['${{matrix.target}}']" >> rust-toolchain.toml
+
       - name: Setup Rust toolchain and cache
         uses: actions-rust-lang/setup-rust-toolchain@v1.4.3
         with:

rust-toolchain.toml

@@ -0,0 +1,2 @@
+[toolchain]
+channel = "nightly"

sdk/couchbase-core/src/kvclientmanager.rs

@@ -239,10 +239,8 @@ where
         let client = manager.get_client(endpoint.clone()).await?;
 
         let res = operation(client.clone()).await;
-        match res {
-            Ok(r) => {
-                return Ok(r);
-            }
+        return match res {
+            Ok(r) => Ok(r),
             Err(e) => {
                 if let Some(memdx_err) = e.is_memdx_error() {
                     if memdx_err.is_dispatch_error() {
@@ -257,20 +255,9 @@ where
                     }
                 }
 
-                return Err(e);
+                Err(e)
             }
-        }
-    }
-}
-
-pub(crate) struct OrchestrateMemdClientAsyncFnMut {}
-
-impl OrchestrateMemdClientAsyncFnMut {
-    async fn call<K, Resp: TryFromClientResponse>(&mut self, client: Arc<K>) -> Result<Resp>
-    where
-        K: KvClient + KvClientOps + PartialEq + Sync + Send + 'static,
-    {
-        todo!()
+        };
     }
 }
 
@@ -357,7 +344,7 @@ mod tests {
         let result = orchestrate_memd_client(
             &manager,
             "192.168.107.128:11210".to_string(),
-            |client: Arc<StdKvClient<Client>>| async move {
+            async |client: Arc<StdKvClient<Client>>| {
                 client
                     .set(SetRequest {
                         collection_id: 0,
@@ -379,7 +366,7 @@ mod tests {
         .await
         .unwrap();
 
-        dbg!(result);
+        // dbg!(result);
 
         let client = manager
             .get_client("192.168.107.128:11210".to_string())

sdk/couchbase-core/src/lib.rs

@@ -1,3 +1,5 @@
+#![feature(async_closure)]
+
 pub mod authenticator;
 pub mod cbconfig;
 mod configparser;

sdk/couchbase-core/src/vbucketrouter.rs

@@ -79,10 +79,10 @@ pub(crate) trait NotMyVbucketConfigHandler {
 
 pub(crate) async fn orchestrate_memd_routing<V, Resp: TryFromClientResponse, Fut>(
     vb: &V,
-    ch: Option<Arc<impl NotMyVbucketConfigHandler>>,
+    nmvb_handler: Arc<impl NotMyVbucketConfigHandler>,
     key: &[u8],
     vb_server_idx: u32,
-    mut operation: impl FnMut(String, u16) -> Fut,
+    mut operation: impl Fn(String, u16) -> Fut,
 ) -> Result<Resp>
 where
     V: VbucketRouter,
@@ -96,12 +96,6 @@ where
             Err(e) => e,
         };
 
-        let ch = ch.clone();
-
-        if ch.is_none() {
-            return Err(err);
-        }
-
         let config = if let Some(memdx_err) = err.is_memdx_error() {
             if memdx_err.is_notmyvbucket_error() {
                 if let Some(config) = memdx_err.has_server_config() {
@@ -134,7 +128,9 @@ where
             }
         };
 
-        ch.unwrap().not_my_vbucket_config(config_json, &endpoint);
+        nmvb_handler
+            .clone()
+            .not_my_vbucket_config(config_json, &endpoint);
 
         let (new_endpoint, new_vb_id) = vb.dispatch_by_key(key, vb_server_idx)?;
         if new_endpoint == endpoint && new_vb_id == vb_id {
@@ -280,16 +276,19 @@ mod tests {
 
         let dispatcher = StdVbucketRouter::new(routing_info, VbucketRouterOptions {});
 
+        // let dispatcher = Arc::new(dispatcher);
+        // let manager = Arc::new(manager);
+
         let set_result = orchestrate_memd_routing(
             &dispatcher,
-            Some(Arc::new(NVMBHandler {})),
+            Arc::new(NVMBHandler {}),
             b"test",
             0,
-            |endpoint: String, vb_id: u16| async {
+            async |endpoint: String, vb_id: u16| {
                 orchestrate_memd_client(
                     &manager,
                     endpoint,
-                    |client: Arc<StdKvClient<Client>>| async move {
+                    async |client: Arc<StdKvClient<Client>>| {
                         client
                             .set(SetRequest {
                                 collection_id: 0,
@@ -318,14 +317,14 @@ mod tests {
 
         let get_result = orchestrate_memd_routing(
             &dispatcher,
-            Some(Arc::new(NVMBHandler {})),
+            Arc::new(NVMBHandler {}),
             b"test",
             0,
-            |endpoint: String, vb_id: u16| async {
+            async |endpoint: String, vb_id: u16| {
                 orchestrate_memd_client(
                     &manager,
                     endpoint,
-                    |client: Arc<StdKvClient<Client>>| async move {
+                    async |client: Arc<StdKvClient<Client>>| {
                         client
                             .get(GetRequest {
                                 collection_id: 0,