temporalio · Sushisource · Nov 6, 2024 · Oct 25, 2024 · Oct 28, 2024 · Oct 28, 2024
@@ -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"

@@ -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"),
         }
     }
+1 −0		core-api/Cargo.toml
+84 −30		core-api/src/worker.rs
+94 −10		core/src/abstractions.rs
+4 −3		core/src/core_tests/workflow_tasks.rs
+6 −6		core/src/pollers/poll_buffer.rs
+8 −8		core/src/worker/activities.rs
+19 −6		core/src/worker/activities/local_activities.rs
+16 −6		core/src/worker/mod.rs
+5 −7		core/src/worker/slot_provider.rs
+18 −6		core/src/worker/tuner.rs
+4 −3		core/src/worker/tuner/fixed_size.rs
+159 −29		core/src/worker/tuner/resource_based.rs
+2 −2		core/src/worker/workflow/mod.rs
+4 −3		core/src/worker/workflow/wft_extraction.rs
+16 −0		sdk-core-protos/protos/local/temporal/sdk/core/core_interface.proto
+44 −8		tests/integ_tests/worker_tests.rs