1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
/*
* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at https://mozilla.org/MPL/2.0/.
*/
//! This module contains the basic block of [`fetcher`](`crate`) that is a [`Task`].
mod disabled_task;
mod opaque_task;
mod task_group;
pub mod entry_to_msg_map;
use std::convert::Infallible;
pub use self::disabled_task::DisabledTask;
pub use self::opaque_task::OpaqueTask;
pub use self::task_group::TaskGroup;
use self::entry_to_msg_map::EntryToMsgMap;
use crate::{
StaticStr,
actions::{Action, ActionContext, ActionResult},
cancellation_token::CancellationToken,
entry::Entry,
error::FetcherError,
external_save::ExternalSave,
sources::Source,
};
/// A core primitive of [`fetcher`](`crate`). A single instance of a data pipeline.
///
/// Runs the data fetched from a [`Source`] through the pipeline ([`Task::action`])
#[derive(bon::Builder, Clone, Debug)]
#[non_exhaustive]
pub struct Task<S, A, E> {
/// Name of the task
#[builder(start_fn, into)]
pub name: StaticStr,
/// Map of an entry (by [`EntryId`](`crate::entry::EntryId`)) to a sent message (by [`MessageId`](`crate::sinks::message::MessageId`)).
///
/// Sinks supporting replies can make the current message a reply to an older one.
pub entry_to_msg_map: Option<EntryToMsgMap<E>>,
/// Optional tag that a [`Sink`](`crate::sinks::Sink`) may put near a message body to differentiate this task from others that may be similar.
///
/// For example, messages from different task that are sent to the same sink can be differentiated using this adjecent tag.
#[builder(into)]
pub tag: Option<StaticStr>,
/// Source where to fetch the data from.
///
/// Also used to mark the entry as read after it's been sent.
pub source: Option<S>,
/// Pipeline (in other words, a list of actions) which the data received from the source is run through
pub action: Option<A>,
/// Gracefully stop the task mid-work when signalled
///
/// Can be specified when building a [`Job`](`crate::job::Job`)
/// using [`JobBuilder::cancel_token`](`crate::job::JobBuilder::cancel_token`) in which this task is included.
/// The job will propagate the token to all children tasks automatically.
pub cancel_token: Option<CancellationToken>,
}
impl<S, A, E> Task<S, A, E>
where
S: Source,
A: Action,
E: ExternalSave,
{
/// Run a task once to completion
///
/// # Errors
/// Errors if any part of the pipeline (source -> actions) failed,
/// if the [`ReadFilter`](`crate::read_filter::ReadFilter`) failed,
/// or if the [`ExternalSave`] implementation caused the [`EntryToMsgMap`] to return an error.
#[expect(clippy::same_name_method, reason = "can't think of a better name")] // if any come up, I'd be fine to replace it
#[tracing::instrument(skip(self), fields(name = %self.name))]
pub async fn run(&mut self) -> Result<(), FetcherError> {
tracing::trace!("Running task");
let raw = match &mut self.source {
Some(source) => source.fetch().await.map_err(Into::into)?,
None => vec![Entry::default()], // return just an empty entry if there is no source
};
tracing::debug!("Got {} raw entries from the sources", raw.len());
tracing::trace!("Raw entries: {raw:#?}");
if let Some(action) = &mut self.action {
let ctx = ActionContext {
source: self.source.as_mut(),
entry_to_msg_map: self.entry_to_msg_map.as_mut(),
tag: self.tag.as_deref(),
cancel_token: self.cancel_token.as_ref(),
};
match action.apply(raw, ctx).await {
ActionResult::Ok(_) | ActionResult::Terminated => (),
ActionResult::Err(e) => return Err(e.into()),
}
}
Ok(())
}
}
impl<S, A, E> OpaqueTask for Task<S, A, E>
where
S: Source,
A: Action,
E: ExternalSave,
{
async fn run(&mut self) -> Result<(), FetcherError> {
Task::run(self).await
}
fn set_cancel_token(&mut self, channel: CancellationToken) {
self.cancel_token = Some(channel);
}
}
impl<S, A, State: task_builder::State> TaskBuilder<S, A, Infallible, State> {
/// Disables [`Task::entry_to_msg_map`].
///
/// Even though [`Task::entry_to_msg_map`] is optional, the generic still needs to be specified.
/// This method specifies the generic as [`()`] and sets [`Task::entry_to_msg_map`] to `None`.
pub fn no_entry_to_msg_map(
self,
) -> TaskBuilder<S, A, Infallible, task_builder::SetEntryToMsgMap<State>>
where
State::EntryToMsgMap: task_builder::IsUnset,
{
self.maybe_entry_to_msg_map(None::<EntryToMsgMap<Infallible>>)
}
/// Builds the task while disabling the [`Task::entry_to_msg_map`] via [`TaskBuilder::no_entry_to_msg_map`].
pub fn build_without_replies(self) -> Task<S, A, Infallible>
where
State::EntryToMsgMap: task_builder::IsUnset,
{
self.no_entry_to_msg_map().build()
}
}