wp-core-connectors 0.3.0

Core connector registry and sink runtimes for WarpParse
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 super::blackhole::BlackHoleSink;
use async_trait::async_trait;
use orion_error::conversion::ToStructError;
use wp_connector_api::{
    ConnectorDef, ParamMap, SinkBuildCtx, SinkDefProvider, SinkFactory, SinkHandle, SinkReason,
    SinkResult,
};

pub struct BlackHoleFactory;

struct BlackHoleSpec {
    sleep_ms: u64,
}

impl BlackHoleSpec {
    fn from_params(params: &ParamMap) -> SinkResult<Self> {
        if let Some(value) = params.get("sleep_ms")
            && value.as_u64().is_none()
        {
            return Err(SinkReason::core_conf()
                .to_err()
                .with_detail("blackhole.sleep_ms must be an unsigned integer"));
        }
        let sleep_ms = params.get("sleep_ms").and_then(|v| v.as_u64()).unwrap_or(0);
        Ok(Self { sleep_ms })
    }
}

#[async_trait]
impl SinkFactory for BlackHoleFactory {
    fn kind(&self) -> &'static str {
        "blackhole"
    }
    fn validate_spec(&self, spec: &wp_connector_api::SinkSpec) -> SinkResult<()> {
        BlackHoleSpec::from_params(&spec.params)?;
        Ok(())
    }
    async fn build(
        &self,
        spec: &wp_connector_api::SinkSpec,
        _ctx: &SinkBuildCtx,
    ) -> SinkResult<SinkHandle> {
        let resolved = BlackHoleSpec::from_params(&spec.params)?;
        Ok(SinkHandle::new(Box::new(BlackHoleSink::new(
            resolved.sleep_ms,
        ))))
    }
}

impl SinkDefProvider for BlackHoleFactory {
    fn sink_def(&self) -> ConnectorDef {
        crate::builtin::sink_def("blackhole_sink")
            .expect("builtin sink def missing: blackhole_sink")
    }
}