wp-core-connectors 0.7.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
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

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")
                    })?;
                // Validate data_format if provided
                if let Some(df) = spec.params.get("data_format").and_then(|v| v.as_str())
                    && !matches!(df, "arrow_framed" | "arrow_ipc")
                {
                    return Err(SinkReason::core_conf().to_err().with_detail(format!(
                        "file_arrow.data_format must be 'arrow_framed' or 'arrow_ipc' (got '{df}')"
                    )));
                }
                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);
                let data_format = spec
                    .params
                    .get("data_format")
                    .and_then(|v| v.as_str())
                    .unwrap_or("arrow_ipc")
                    .to_ascii_lowercase();
                let framed = data_format == "arrow_framed";
                let tag = spec
                    .params
                    .get("tag")
                    .and_then(|v| v.as_str())
                    .unwrap_or("");
                Box::new(ArrowFileSink::with_format(&file_path, sync, framed, tag)?)
            }
            "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())
    }
}