wf_connector_api/
lib.rs

1//! # wf-connector-api
2//!
3//! Minimal Arrow-native connector API for warp-fusion.
4//!
5//! ## Design
6//!
7//! `wp-connector-api` sources produce `SourceEvent { payload: RawData }`,
8//! designed for downstream parse pipelines. CEP engines like warp-fusion
9//! operate on Arrow `RecordBatch` directly.
10//!
11//! `wf-connector-api` fills this gap for Arrow-native source consumption.
12//! (Sink output uses the existing `wp-connector-api` `SinkRuntime` —
13//! adding `send_batch()` to it is sufficient, no new trait needed.)
14//!
15//! ## Relationship with `wp-connector-api`
16//!
17//! | | wp-connector-api | wf-connector-api |
18//! |---|---|---|
19//! | Source data | `SourceEvent { payload: RawData }` | `RecordBatch` (columnar) |
20//! | Consumer | parse pipeline (WPL) | CEP engine (warp-fusion) |
21//! | Error model | `SourceResult<T>` (orion-error) | `SourceResult<T>` (orion-error) |
22//! | Lifecycle | `start()` / `receive()` / `close()` | `start()` / `receive_batch()` / `close()` |
23//!
24//! `wp-connectors` (the implementation crate) can implement BOTH traits
25//! for the same connector (Kafka / File / TCP), sharing connection logic.
26
27use arrow::record_batch::RecordBatch;
28use async_trait::async_trait;
29use orion_error::conversion::ToStructError;
30use orion_error::{OrionError, StructError, UnifiedReason};
31use std::error::Error as StdError;
32
33// -- Error -------------------------------------------------------------------
34
35/// Connector error reason.
36///
37/// All leaf variants carry detail via `err_detail()`. `SourceError` wraps
38/// each variant with a detail string and optional source error.
39#[derive(Debug, Clone, PartialEq, OrionError)]
40pub enum SourceReason {
41    /// End of stream — no more data will be produced.
42    #[orion_error(message = "end of stream", identity = "sys.wf_connector.eof")]
43    EOF,
44    /// No data currently available (not EOF); caller should retry.
45    #[orion_error(message = "no data available", identity = "sys.wf_connector.not_data")]
46    NotData,
47    /// I/O error from the underlying transport.
48    #[orion_error(message = "I/O error", identity = "sys.wf_connector.io")]
49    Io,
50    /// Failed to establish connection / bind / subscribe.
51    #[orion_error(message = "connection error", identity = "sys.wf_connector.connect")]
52    Connect,
53    /// Message / frame decoding failed.
54    #[orion_error(message = "decode error", identity = "sys.wf_connector.decode")]
55    Decode,
56    /// Referenced connector not found in registry.
57    #[orion_error(
58        message = "connector not found",
59        identity = "sys.wf_connector.not_found"
60    )]
61    NotFound,
62    /// Catch-all for unexpected errors.
63    #[orion_error(transparent)]
64    General(UnifiedReason),
65}
66
67impl SourceReason {
68    /// Create an error with detail message.
69    pub fn err_detail<S: Into<String>>(self, detail: S) -> SourceError {
70        self.to_err().with_detail(detail.into())
71    }
72
73    /// Create an error with a source (chained) error.
74    pub fn err_source<E>(self, source: E) -> SourceError
75    where
76        E: StdError + Send + Sync + 'static,
77    {
78        self.to_err().with_source(source)
79    }
80}
81
82pub type SourceError = StructError<SourceReason>;
83pub type SourceResult<T> = Result<T, SourceError>;
84
85// -- Source ------------------------------------------------------------------
86
87/// A batch-oriented data source that produces Arrow [`RecordBatch`]es.
88///
89/// # Lifecycle
90///
91/// 1. `start()` — initialize (connect, subscribe, bind)
92/// 2. `receive_batch()` — pull data in a loop
93/// 3. `close()` — release resources (unsubscribe, close connections)
94///
95/// `close()` must be idempotent — safe to call multiple times, even before `start()`.
96///
97/// # Empty vs EOF
98///
99/// - Return `Ok(vec![])` when no data is currently available (caller should retry).
100/// - Return `Err(SourceReason::EOF.into())` when the stream has ended.
101#[async_trait]
102pub trait BatchSource: Send {
103    /// Initialize the source. Called once before the first `receive_batch()`.
104    ///
105    /// Default is a no-op.
106    async fn start(&mut self) -> SourceResult<()> {
107        Ok(())
108    }
109
110    /// Receive zero or more [`RecordBatch`]es.
111    ///
112    /// An empty `Vec` means "no data right now" — the caller should poll again.
113    /// An error with `SourceReason::EOF` means the stream has ended.
114    async fn receive_batch(&mut self) -> SourceResult<Vec<RecordBatch>>;
115
116    /// Close the source and release all resources.
117    ///
118    /// Must be idempotent — safe to call multiple times or before `start()`.
119    /// Default is a no-op.
120    async fn close(&mut self) -> SourceResult<()> {
121        Ok(())
122    }
123
124    /// Unique identifier for this source instance (logging / metrics).
125    fn identifier(&self) -> &str;
126}
127
128// -- Sink (not needed as a separate trait) -----------------------------------
129//
130// Arrow-native sink output is handled by the existing `wp-connector-api`
131// `SinkRuntime`. Adding a `send_batch()` method to `SinkRuntime` (which
132// already has `send_record()`) is sufficient — no new trait required.
133// File / Arrow IPC / TCP backends can natively accept RecordBatch.
wf_connector_api/lib.rs

wf_connector_api/
lib.rs
