wp-core-connectors 0.5.2

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
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

use async_trait::async_trait;
use orion_error::conversion::{SourceErr, ToStructError};
use wp_connector_api::{
    ConnectorDef, SinkBuildCtx, SinkDefProvider, SinkFactory, SinkHandle, SinkReason, SinkResult,
    SinkSpec,
};

use super::file::{ArrowFileSink, AsyncFileSink, FileSinkSpec, FormattedFileSink};

/// Resolve the output file path from sink spec params.
///
/// Uses `base` (default `"./data/out_dat"`) and `file` (default `"out.dat"`),
/// then resolves relative to the `SinkBuildCtx::work_root`.
fn resolve_file_path(spec: &SinkSpec, ctx: &SinkBuildCtx) -> SinkResult<String> {
    let base = spec
        .params
        .get("base")
        .and_then(|v| v.as_str())
        .unwrap_or("./data/out_dat");
    let file_name = spec
        .params
        .get("file")
        .and_then(|v| v.as_str())
        .unwrap_or("out.dat");
    Ok(super::file::resolve_output_path(base, file_name, ctx)
        .display()
        .to_string())
}

pub struct FileFactory;

#[async_trait]
impl SinkFactory for FileFactory {
    fn kind(&self) -> &'static str {
        "file"
    }

    fn validate_spec(&self, spec: &SinkSpec) -> SinkResult<()> {
        let protocol = spec
            .params
            .get("protocol")
            .and_then(|v| v.as_str())
            .unwrap_or("txt");
        match protocol {
            "arrow" => {
                // Ensure 'file' param is present
                let _ = spec
                    .params
                    .get("file")
                    .and_then(|v| v.as_str())
                    .ok_or_else(|| {
                        SinkReason::core_conf()
                            .to_err()
                            .with_detail("arrow file sink requires 'file' param")
                    })?;
                Ok(())
            }
            "txt" | "" => {
                // Existing text-based validation
                FileSinkSpec::from_resolved("file", spec)?;
                Ok(())
            }
            other => Err(SinkReason::core_conf().to_err().with_detail(format!(
                "unsupported file protocol: '{other}'; expected 'txt' or 'arrow'"
            ))),
        }
    }

    async fn build(&self, spec: &SinkSpec, ctx: &SinkBuildCtx) -> SinkResult<SinkHandle> {
        let protocol = spec
            .params
            .get("protocol")
            .and_then(|v| v.as_str())
            .unwrap_or("txt");

        let sink: Box<dyn wp_connector_api::AsyncSink> = match protocol {
            "arrow" => {
                let file_path = resolve_file_path(spec, ctx)?;
                let sync = spec
                    .params
                    .get("sync")
                    .and_then(|v| v.as_bool())
                    .unwrap_or(false);
                Box::new(ArrowFileSink::new(&file_path, sync)?)
            }
            "txt" | "" => {
                let resolved = FileSinkSpec::from_resolved("file", spec)?;
                let path = resolved.resolve_path(ctx);
                let fmt = resolved.text_fmt();
                let sync = resolved.sync();
                let sink = AsyncFileSink::with_sync(&path, sync)
                    .await
                    .source_err(SinkReason::Sink, "file sink open")?;
                Box::new(FormattedFileSink::new(fmt, sink))
            }
            other => {
                return Err(SinkReason::core_conf().to_err().with_detail(format!(
                    "unsupported file protocol: '{other}'; expected 'txt' or 'arrow'"
                )));
            }
        };

        Ok(SinkHandle::new(sink))
    }
}

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

    fn sink_defs(&self) -> Vec<ConnectorDef> {
        crate::builtin::sink_defs_by_kind(self.kind())
    }
}