very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust
fjall
at-protocol
atproto
indexer
[lib] make Hydrant::run return a result instead of panicking when run more than once
+11
-11
+1
-1
examples/statusphere.rs
+1
-1
examples/statusphere.rs
···
189
189
let index = Arc::new(StatusIndex::new());
190
190
tokio::select! {
191
191
// this finally starts hydrant, so it will start crawling and backfilling etc.
192
-
r = hydrant.run() => r,
192
+
r = hydrant.run()? => r,
193
193
_ = run_ticker(index.clone()) => Ok(()),
194
194
_ = handle_stream(index.clone(), hydrant.repos.clone(), stream) => Ok(()),
195
195
}
+8
-8
src/control.rs
+8
-8
src/control.rs
···
162
162
/// resolves with `Ok(())` if a fatal component exits cleanly, or `Err(e)` if it
163
163
/// fails. intended for use in `tokio::select!` alongside [`serve`](Self::serve).
164
164
///
165
-
/// panics if called more than once on the same `Hydrant` instance.
166
-
pub fn run(&self) -> impl Future<Output = Result<()>> {
165
+
/// returns an error if called more than once on the same `Hydrant` instance.
166
+
pub fn run(&self) -> Result<impl Future<Output = Result<()>>> {
167
167
let state = self.state.clone();
168
168
let config = self.config.clone();
169
-
let started = self.started.clone();
170
169
171
-
async move {
172
-
if started.swap(true, Ordering::SeqCst) {
173
-
panic!("Hydrant::run() called more than once");
174
-
}
170
+
if self.started.swap(true, Ordering::SeqCst) {
171
+
miette::bail!("Hydrant::run() called more than once");
172
+
}
175
173
174
+
let fut = async move {
176
175
// internal buffered channel between ingestors / backfill and the firehose worker
177
176
let (buffer_tx, buffer_rx) = mpsc::unbounded_channel();
178
177
···
499
498
Err(_) => return Ok(()),
500
499
}
501
500
}
502
-
}
501
+
};
502
+
Ok(fut)
503
503
}
504
504
505
505
/// subscribe to the ordered event stream.
+2
-2
src/main.rs
+2
-2
src/main.rs
···
23
23
24
24
if app.enable_debug {
25
25
tokio::select! {
26
-
r = hydrant.run() => r,
26
+
r = hydrant.run()? => r,
27
27
r = hydrant.serve(app.api_port) => r,
28
28
r = hydrant.serve_debug(app.debug_port) => r,
29
29
}
30
30
} else {
31
31
tokio::select! {
32
-
r = hydrant.run() => r,
32
+
r = hydrant.run()? => r,
33
33
r = hydrant.serve(app.api_port) => r,
34
34
}
35
35
}