fission_core/
async_runtime.rs

1use crate::action::ActionEnvelope;
2use crate::data_stream::{
3    BoxFissionDataStream, DataStreamId, DataStreamRegistry, FissionDataStreamError,
4};
5use serde::{de::DeserializeOwned, Deserialize, Serialize};
6use std::borrow::Cow;
7use std::future::Future;
8use std::marker::PhantomData;
9use std::pin::Pin;
10use std::sync::Arc;
11
12pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
13
14pub trait JobSpec {
15    type Request: Serialize + DeserializeOwned + Send + 'static;
16    type Ok: Serialize + DeserializeOwned + Send + 'static;
17    type Err: Serialize + DeserializeOwned + Send + 'static;
18    const NAME: &'static str;
19}
20
21#[derive(Debug)]
22pub struct JobRef<J: JobSpec> {
23    pub name: &'static str,
24    _marker: PhantomData<fn() -> J>,
25}
26
27impl<J: JobSpec> JobRef<J> {
28    pub const fn new(name: &'static str) -> Self {
29        Self {
30            name,
31            _marker: PhantomData,
32        }
33    }
34}
35
36impl<J: JobSpec> Clone for JobRef<J> {
37    fn clone(&self) -> Self {
38        *self
39    }
40}
41
42impl<J: JobSpec> Copy for JobRef<J> {}
43
44pub trait ServiceSpec {
45    type Config: Serialize + DeserializeOwned + Send + 'static;
46    type Command: Serialize + DeserializeOwned + Send + 'static;
47    type CommandOk: Serialize + DeserializeOwned + Send + 'static;
48    type CommandErr: Serialize + DeserializeOwned + Send + 'static;
49    type Event: Serialize + DeserializeOwned + Send + 'static;
50    type StartErr: Serialize + DeserializeOwned + Send + 'static;
51    const NAME: &'static str;
52}
53
54#[derive(Debug)]
55pub struct ServiceType<S: ServiceSpec> {
56    pub name: &'static str,
57    _marker: PhantomData<fn() -> S>,
58}
59
60impl<S: ServiceSpec> ServiceType<S> {
61    pub const fn new(name: &'static str) -> Self {
62        Self {
63            name,
64            _marker: PhantomData,
65        }
66    }
67}
68
69impl<S: ServiceSpec> Clone for ServiceType<S> {
70    fn clone(&self) -> Self {
71        *self
72    }
73}
74
75impl<S: ServiceSpec> Copy for ServiceType<S> {}
76
77#[derive(Debug)]
78pub struct ServiceSlot<S: ServiceSpec> {
79    pub ty: ServiceType<S>,
80    pub slot_key: Cow<'static, str>,
81}
82
83impl<S: ServiceSpec> ServiceSlot<S> {
84    pub fn singleton(ty: ServiceType<S>) -> Self {
85        Self {
86            ty,
87            slot_key: Cow::Borrowed("singleton"),
88        }
89    }
90
91    pub fn keyed(ty: ServiceType<S>, key: impl Into<String>) -> Self {
92        Self {
93            ty,
94            slot_key: Cow::Owned(key.into()),
95        }
96    }
97
98    pub fn slot_key(&self) -> &str {
99        self.slot_key.as_ref()
100    }
101}
102
103impl<S: ServiceSpec> Clone for ServiceSlot<S> {
104    fn clone(&self) -> Self {
105        Self {
106            ty: self.ty,
107            slot_key: self.slot_key.clone(),
108        }
109    }
110}
111
112#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
113pub struct JobRequestPayload {
114    pub job_name: String,
115    pub payload: Vec<u8>,
116}
117
118#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
119pub struct ServiceStartPayload {
120    pub service_name: String,
121    pub slot_key: String,
122    pub config: Vec<u8>,
123}
124
125#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
126pub struct ServiceCommandPayload {
127    pub service_name: String,
128    pub slot_key: String,
129    pub payload: Vec<u8>,
130}
131
132#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
133pub struct ServiceStopPayload {
134    pub service_name: String,
135    pub slot_key: String,
136}
137
138#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Default)]
139pub struct ServiceBindings {
140    pub on_started: Option<ActionEnvelope>,
141    pub on_start_failed: Option<ActionEnvelope>,
142    pub on_event: Option<ActionEnvelope>,
143    pub on_stopped: Option<ActionEnvelope>,
144    pub on_command_ok: Option<ActionEnvelope>,
145    pub on_command_err: Option<ActionEnvelope>,
146}
147
148#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
149pub struct ResourceExecutionContext {
150    pub key: String,
151    pub generation: u64,
152}
153
154#[derive(Clone, Debug)]
155pub struct JobCtx {
156    pub req_id: u64,
157    data_streams: DataStreamRegistry,
158}
159
160impl JobCtx {
161    #[doc(hidden)]
162    pub fn new_runtime(req_id: u64, data_streams: DataStreamRegistry) -> Self {
163        Self {
164            req_id,
165            data_streams,
166        }
167    }
168
169    /// Opens a runtime-owned data stream for this job.
170    ///
171    /// Streams are one-shot by default; opening consumes the stream handle.
172    pub fn open_data_stream(
173        &self,
174        id: DataStreamId,
175    ) -> Result<BoxFissionDataStream, FissionDataStreamError> {
176        self.data_streams.open(id)
177    }
178
179    /// Releases a stream without consuming it.
180    pub fn release_data_stream(&self, id: DataStreamId) -> bool {
181        self.data_streams.release(id)
182    }
183}
184
185type EmitFn = dyn Fn(Vec<u8>) -> BoxFuture<Result<(), String>> + Send + Sync;
186
187struct ServiceCtxInner {
188    service_name: String,
189    slot_key: String,
190    instance_id: u64,
191    data_streams: DataStreamRegistry,
192    emit: Arc<EmitFn>,
193}
194
195pub struct ServiceCtx<S: ServiceSpec> {
196    inner: Arc<ServiceCtxInner>,
197    _marker: PhantomData<fn() -> S>,
198}
199
200impl<S: ServiceSpec> std::fmt::Debug for ServiceCtx<S> {
201    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
202        f.debug_struct("ServiceCtx")
203            .field("service_name", &self.inner.service_name)
204            .field("slot_key", &self.inner.slot_key)
205            .field("instance_id", &self.inner.instance_id)
206            .finish()
207    }
208}
209
210impl<S: ServiceSpec> Clone for ServiceCtx<S> {
211    fn clone(&self) -> Self {
212        Self {
213            inner: self.inner.clone(),
214            _marker: PhantomData,
215        }
216    }
217}
218
219impl<S: ServiceSpec> ServiceCtx<S> {
220    #[doc(hidden)]
221    pub fn new_runtime(
222        service_name: String,
223        slot_key: String,
224        instance_id: u64,
225        data_streams: DataStreamRegistry,
226        emit: Arc<EmitFn>,
227    ) -> Self {
228        Self {
229            inner: Arc::new(ServiceCtxInner {
230                service_name,
231                slot_key,
232                instance_id,
233                data_streams,
234                emit,
235            }),
236            _marker: PhantomData,
237        }
238    }
239
240    pub fn service_name(&self) -> &str {
241        &self.inner.service_name
242    }
243
244    pub fn slot_key(&self) -> &str {
245        &self.inner.slot_key
246    }
247
248    pub fn instance_id(&self) -> u64 {
249        self.inner.instance_id
250    }
251
252    pub fn emit(&self, event: S::Event) -> BoxFuture<Result<(), String>> {
253        match serde_json::to_vec(&event) {
254            Ok(bytes) => (self.inner.emit)(bytes),
255            Err(err) => Box::pin(async move { Err(err.to_string()) }),
256        }
257    }
258
259    /// Opens a runtime-owned data stream for this service.
260    ///
261    /// Streams are one-shot by default; opening consumes the stream handle.
262    pub fn open_data_stream(
263        &self,
264        id: DataStreamId,
265    ) -> Result<BoxFissionDataStream, FissionDataStreamError> {
266        self.inner.data_streams.open(id)
267    }
268
269    /// Releases a stream without consuming it.
270    pub fn release_data_stream(&self, id: DataStreamId) -> bool {
271        self.inner.data_streams.release(id)
272    }
273}
274
275pub trait ServiceRunner<S: ServiceSpec>: Send + 'static {
276    fn on_command(
277        &mut self,
278        command: S::Command,
279        ctx: ServiceCtx<S>,
280    ) -> BoxFuture<Result<S::CommandOk, S::CommandErr>>;
281
282    fn on_stop(self: Box<Self>, ctx: ServiceCtx<S>) -> BoxFuture<()>;
283}
fission_core/async_runtime.rs

fission_core/
async_runtime.rs
