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
148
149
150
151
152
pub mod entry_to_msg_map;
use std::collections::HashSet;
use self::entry_to_msg_map::EntryToMsgMap;
use crate::{
action::{transform::error::TransformError, Action},
entry::Entry,
error::Error,
sink::Sink,
source::Source,
};
#[derive(Debug)]
pub struct Task {
pub tag: Option<String>,
pub source: Option<Box<dyn Source>>,
pub actions: Option<Vec<Action>>,
pub sink: Option<Box<dyn Sink>>,
pub entry_to_msg_map: Option<EntryToMsgMap>,
}
impl Task {
#[tracing::instrument(skip(self))]
pub async fn run(&mut self) -> Result<(), Error> {
tracing::trace!("Running task");
let entries = {
let raw = match &mut self.source {
Some(source) => source.fetch().await?,
None => vec![Entry::default()], };
tracing::debug!("Got {} raw entries from the source(s)", raw.len());
let processed = match &self.actions {
Some(actions) => process_entries(raw, actions).await?,
None => raw,
};
tracing::debug!("Got {} fully processed entries", processed.len());
remove_duplicates(processed)
};
for entry in entries.into_iter().rev() {
let msgid = match self.sink.as_ref() {
Some(sink) if !entry.msg.is_empty() || entry.raw_contents.is_some() => {
let mut msg = entry.msg;
if msg.is_empty() {
msg.body = Some(
entry
.raw_contents
.expect("raw_contents should be some because of the match guard"),
);
}
let tag = self.tag.as_deref();
let reply_to = self
.entry_to_msg_map
.as_mut()
.and_then(|map| map.get_if_exists(entry.id.as_ref()));
tracing::debug!(
"Sending {msg:?} to a sink with tag {tag:?}, replying to {reply_to:?}"
);
sink.send(msg, reply_to, tag).await?
}
_ => None,
};
if let Some(entry_id) = entry.id {
if let Some(source) = &mut self.source {
tracing::debug!("Marking {entry_id:?} as read");
source.mark_as_read(&entry_id).await?;
}
if let Some((msgid, map)) = msgid.zip(self.entry_to_msg_map.as_mut()) {
tracing::debug!("Associating entry {entry_id:?} with message {msgid:?}");
map.insert(entry_id, msgid).await?;
}
}
}
Ok(())
}
}
async fn process_entries(
mut entries: Vec<Entry>,
actions: &[Action],
) -> Result<Vec<Entry>, TransformError> {
for a in actions {
entries = a.process(entries).await?;
}
Ok(entries)
}
fn remove_duplicates(entries: Vec<Entry>) -> Vec<Entry> {
let num_og_entries = entries.len();
let mut uniq = Vec::new();
let mut used_ids = HashSet::new();
for ent in entries {
match ent.id.as_deref() {
Some("") => panic!("An id should never be none but empty"),
Some(id) => {
if used_ids.insert(id.to_owned()) {
uniq.push(ent);
}
}
None => uniq.push(ent),
}
}
let num_removed = num_og_entries - uniq.len();
if num_removed > 0 {
tracing::trace!("Removed {} duplicate entries", num_removed);
}
uniq
}