exc 0.7.3

The abstraction layer of exchanges.
Documentation 
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

use std::{
    sync::Arc,
    task::{Context, Poll},
};

use exc_core::ExchangeError;
use futures::{future::BoxFuture, FutureExt};

use super::InstrumentsOptions;

use super::{state::State, FetchInstrumentSvc, SubscribeInstrumentSvc};

pub(super) struct Worker {
    init: Option<BoxFuture<'static, Result<(), ExchangeError>>>,
    state: Arc<State>,
    inst: SubscribeInstrumentSvc,
    opts: InstrumentsOptions,
}

impl Worker {
    pub(super) fn new(
        state: &Arc<State>,
        opts: &InstrumentsOptions,
        inst: SubscribeInstrumentSvc,
        fetch: FetchInstrumentSvc,
    ) -> Self {
        let init = state.clone().init(fetch, opts.inst_tags.clone()).boxed();
        Self {
            init: Some(init),
            state: state.clone(),
            inst,
            opts: opts.clone(),
        }
    }

    pub(super) fn poll_init(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ExchangeError>> {
        let Some(fut) = self.init.as_mut() else {
            return Poll::Ready(Ok(()));
        };
        fut.poll_unpin(cx)
    }

    pub(super) async fn start(self) -> Result<(), ExchangeError> {
        let Self {
            state, inst, opts, ..
        } = self;
        let inst = state.watch_instruments(inst, opts.inst_tags);
        tokio::select! {
            res = inst => {
                res?;
            }
        }
        Ok(())
    }
}