User-implementable SlotSupplier (#1553)

Author: Sushisource

packages/core-bridge/Cargo.lock

@@ -17,9 +17,10 @@ lto = true
 incremental = false
 
 [dependencies]
+async-trait = "0.1.83"
 futures = { version = "0.3", features = ["executor"] }
 log = "0.4"
-neon = { version = "1.0.0", default-features = false, features = ["napi-6"] }
+neon = { version = "1.0.0", default-features = false, features = ["napi-6", "futures"] }
 opentelemetry = "0.24"
 parking_lot = "0.12"
 prost = "0.13"

packages/core-bridge/Cargo.toml

@@ -5,8 +5,10 @@ use neon::{
     prelude::*,
     types::{JsBoolean, JsNumber, JsString},
 };
+use slot_supplier_bridge::SlotSupplierBridge;
 use std::{collections::HashMap, net::SocketAddr, sync::Arc, time::Duration};
 use temporal_client::HttpConnectProxyOptions;
+use temporal_sdk_core::api::worker::SlotKind;
 use temporal_sdk_core::{
     api::telemetry::{Logger, MetricTemporality, TelemetryOptions, TelemetryOptionsBuilder},
     api::{
@@ -25,6 +27,8 @@ use temporal_sdk_core::{
     TlsConfig, TunerHolderOptionsBuilder, Url,
 };
 
+mod slot_supplier_bridge;
+
 pub enum EphemeralServerConfig {
     TestServer(TestServerConfig),
     DevServer(TemporalDevServerConfig),
@@ -65,11 +69,11 @@ pub(crate) trait ObjectHandleConversionsExt {
         &self,
         cx: &mut FunctionContext,
     ) -> NeonResult<HashMap<String, String>>;
-    fn as_slot_supplier(
-        &self,
+    fn into_slot_supplier<SK: SlotKind + Send + Sync + 'static>(
+        self,
         cx: &mut FunctionContext,
         rbo: &mut Option<ResourceBasedSlotsOptions>,
-    ) -> NeonResult<SlotSupplierOptions>;
+    ) -> NeonResult<SlotSupplierOptions<SK>>;
 }
 
 impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
@@ -409,18 +413,18 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
             if let Some(wf_slot_supp) =
                 js_optional_getter!(cx, &tuner, "workflowTaskSlotSupplier", JsObject)
             {
-                tuner_holder.workflow_slot_options(wf_slot_supp.as_slot_supplier(cx, &mut rbo)?);
+                tuner_holder.workflow_slot_options(wf_slot_supp.into_slot_supplier(cx, &mut rbo)?);
             }
             if let Some(act_slot_supp) =
                 js_optional_getter!(cx, &tuner, "activityTaskSlotSupplier", JsObject)
             {
-                tuner_holder.activity_slot_options(act_slot_supp.as_slot_supplier(cx, &mut rbo)?);
+                tuner_holder.activity_slot_options(act_slot_supp.into_slot_supplier(cx, &mut rbo)?);
             }
             if let Some(local_act_slot_supp) =
                 js_optional_getter!(cx, &tuner, "localActivityTaskSlotSupplier", JsObject)
             {
                 tuner_holder.local_activity_slot_options(
-                    local_act_slot_supp.as_slot_supplier(cx, &mut rbo)?,
+                    local_act_slot_supp.into_slot_supplier(cx, &mut rbo)?,
                 );
             }
             if let Some(rbo) = rbo {
@@ -567,20 +571,20 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
         }
     }
 
-    fn as_slot_supplier(
-        &self,
+    fn into_slot_supplier<SK: SlotKind + Send + Sync + 'static>(
+        self,
         cx: &mut FunctionContext,
         rbo: &mut Option<ResourceBasedSlotsOptions>,
-    ) -> NeonResult<SlotSupplierOptions> {
-        match js_value_getter!(cx, self, "type", JsString).as_str() {
+    ) -> NeonResult<SlotSupplierOptions<SK>> {
+        match js_value_getter!(cx, &self, "type", JsString).as_str() {
             "fixed-size" => Ok(SlotSupplierOptions::FixedSize {
-                slots: js_value_getter!(cx, self, "numSlots", JsNumber) as usize,
+                slots: js_value_getter!(cx, &self, "numSlots", JsNumber) as usize,
             }),
             "resource-based" => {
-                let min_slots = js_value_getter!(cx, self, "minimumSlots", JsNumber);
-                let max_slots = js_value_getter!(cx, self, "maximumSlots", JsNumber);
-                let ramp_throttle = js_value_getter!(cx, self, "rampThrottleMs", JsNumber) as u64;
-                if let Some(tuner_opts) = js_optional_getter!(cx, self, "tunerOptions", JsObject) {
+                let min_slots = js_value_getter!(cx, &self, "minimumSlots", JsNumber);
+                let max_slots = js_value_getter!(cx, &self, "maximumSlots", JsNumber);
+                let ramp_throttle = js_value_getter!(cx, &self, "rampThrottleMs", JsNumber) as u64;
+                if let Some(tuner_opts) = js_optional_getter!(cx, &self, "tunerOptions", JsObject) {
                     let target_mem =
                         js_value_getter!(cx, &tuner_opts, "targetMemoryUsage", JsNumber);
                     let target_cpu = js_value_getter!(cx, &tuner_opts, "targetCpuUsage", JsNumber);
@@ -603,6 +607,10 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
                     ),
                 ))
             }
+            "custom" => {
+                let ssb = SlotSupplierBridge::new(cx, self)?;
+                Ok(SlotSupplierOptions::Custom(Arc::new(ssb)))
+            }
             _ => cx.throw_type_error("Invalid slot supplier type"),
         }
     }