momento: re-add zincr and zmscore

Re-adds support for sorted set incremenet and sorted set get scores now
that they are in the reworked SDK

Cargo.toml

@@ -26,7 +26,7 @@ hyper = { version = "1.0.0-rc.4", features = ["http1", "http2", "client"]}
 metriken = "0.7.0"
 metriken-exposition = { version = "0.8.0", features = ["json", "parquet-conversion"] }
 mio = "0.8.8"
-momento = "0.39.7"
+momento = "0.40.0"
 pelikan-net = { version = "0.3.0", default-features = false }
 once_cell = "1.18.0"
 openssl = { version = "0.10.64", optional = true }

src/clients/momento/commands/mod.rs

@@ -21,6 +21,7 @@ mod set_add;
 mod set_members;
 mod set_remove;
 mod sorted_set_add;
+mod sorted_set_increment;
 mod sorted_set_range;
 mod sorted_set_rank;
 mod sorted_set_remove;
@@ -45,6 +46,7 @@ pub use set_add::*;
 pub use set_members::*;
 pub use set_remove::*;
 pub use sorted_set_add::*;
+pub use sorted_set_increment::*;
 pub use sorted_set_range::*;
 pub use sorted_set_rank::*;
 pub use sorted_set_remove::*;

src/clients/momento/commands/sorted_set_increment.rs

@@ -0,0 +1,27 @@
+use super::*;
+
+use ::momento::cache::{CollectionTtl, SortedSetIncrementScoreRequest};
+
+/// Increment the score for a member in a sorted set.
+///
+/// NOTE: if a TTL is specified, this command will not refresh the TTL for the
+/// collection.
+pub async fn sorted_set_increment(
+    client: &mut CacheClient,
+    config: &Config,
+    cache_name: &str,
+    request: workload::client::SortedSetIncrement,
+) -> std::result::Result<(), ResponseError> {
+    SORTED_SET_INCR.increment();
+
+    let r = SortedSetIncrementScoreRequest::new(cache_name, &*request.key, &*request.member, request.amount)
+        .ttl(CollectionTtl::new(request.ttl, false));
+
+    let result = timeout(
+        config.client().unwrap().request_timeout(),
+        client.send_request(r),
+    )
+    .await;
+
+    record_result!(result, SORTED_SET_INCR)
+}

src/clients/momento/commands/sorted_set_score.rs

@@ -9,18 +9,25 @@ pub async fn sorted_set_score(
 ) -> std::result::Result<(), ResponseError> {
     if request.members.is_empty() {
         return Ok(());
-    } else if request.members.len() > 1 {
-        REQUEST_UNSUPPORTED.increment();
-        return Ok(());
     }
 
     SORTED_SET_SCORE.increment();
 
-    let result = timeout(
-        config.client().unwrap().request_timeout(),
-        client.sorted_set_get_score(cache_name, &*request.key, &*request.members[0]),
-    )
-    .await;
+    if request.members.len() == 1 {
+        let result = timeout(
+            config.client().unwrap().request_timeout(),
+            client.sorted_set_get_score(cache_name, &*request.key, &*request.members[0]),
+        )
+        .await;
+
+        record_result!(result, SORTED_SET_SCORE)
+    } else {
+        let result = timeout(
+            config.client().unwrap().request_timeout(),
+            client.sorted_set_get_scores(cache_name, &*request.key, request.members.iter().map(|f| &**f)),
+        )
+        .await;
 
-    record_result!(result, SORTED_SET_SCORE)
+        record_result!(result, SORTED_SET_SCORE)
+    }
 }

src/clients/momento/mod.rs

@@ -139,6 +139,9 @@ async fn task(
                 ClientRequest::SortedSetAdd(r) => {
                     sorted_set_add(&mut client, &config, cache_name, r).await
                 }
+                ClientRequest::SortedSetIncrement(r) => {
+                    sorted_set_increment(&mut client, &config, cache_name, r).await
+                }
                 ClientRequest::SortedSetRange(r) => {
                     sorted_set_range(&mut client, &config, cache_name, r).await
                 }