ironflow_worker/api_store/
signal_sender.rs

1//! [`SignalSender`] for [`ApiRunStore`]: on a worker, `ctx.send_signal`
2//! delivers through `POST /api/v1/internal/signals`, where the API's engine
3//! stores the signal, wakes the waiting runs and writes the audit log.
4
5use reqwest::StatusCode;
6use serde::Serialize;
7
8use ironflow_engine::error::EngineError;
9use ironflow_engine::signal::{SignalDelivery, SignalOrigin, SignalSendFuture, SignalSender};
10use ironflow_store::entities::NewSignal;
11
12use super::{ApiResponse, ApiRunStore};
13
14/// Body of `POST /internal/signals`.
15#[derive(Serialize)]
16struct SendSignalBody<'a> {
17    #[serde(flatten)]
18    signal: &'a NewSignal,
19    origin: SignalOrigin,
20}
21
22impl SignalSender for ApiRunStore {
23    fn send(&self, signal: NewSignal, origin: SignalOrigin) -> SignalSendFuture<'_> {
24        Box::pin(async move {
25            let body = SendSignalBody {
26                signal: &signal,
27                origin,
28            };
29            let resp = self
30                .client
31                .post(self.internal("/signals"))
32                .bearer_auth(&self.token)
33                .json(&body)
34                .send()
35                .await
36                .map_err(|e| EngineError::Store(Self::err(e)))?;
37
38            let status = resp.status();
39            if !status.is_success() {
40                let body = resp
41                    .text()
42                    .await
43                    .map_err(|e| EngineError::Store(Self::err(e)))?;
44                return Err(if status == StatusCode::BAD_REQUEST {
45                    EngineError::InvalidSignal(body)
46                } else {
47                    EngineError::Store(Self::status_err(&body))
48                });
49            }
50
51            let api_resp: ApiResponse<SignalDelivery> = resp
52                .json()
53                .await
54                .map_err(|e| EngineError::Store(Self::err(e)))?;
55            Ok(api_resp.data)
56        })
57    }
58}
ironflow_worker/api_store/signal_sender.rs

ironflow_worker/api_store/
signal_sender.rs
