wingfoil 8.0.0

graph based stream processing framework
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

//! Fluvio consumer producer — streams records from a Fluvio topic partition.

use super::{FluvioConnection, FluvioEvent};
use crate::nodes::{RunParams, produce_async};
use crate::types::*;
use fluvio::consumer::ConsumerConfigExt;
use fluvio::{Fluvio, FluvioClusterConfig, Offset};
use futures::StreamExt;
use std::rc::Rc;

/// Stream records from a Fluvio topic partition.
///
/// Connects to the Fluvio SC at `connection.endpoint` and continuously delivers
/// records from `topic` / `partition` as [`FluvioEvent`]s wrapped in a
/// [`Burst`]. Each burst contains exactly one event (records arrive one at a time
/// from the consumer stream).
///
/// # Arguments
///
/// - `connection` — Fluvio cluster configuration (SC endpoint).
/// - `topic` — The Fluvio topic to consume from.
/// - `partition` — Partition index (0-based). Use `0` for single-partition topics.
/// - `start_offset` — Where to begin consuming:
///   - `None` — from the beginning of the partition (`Offset::beginning()`).
///   - `Some(n)` — from absolute offset `n` (inclusive). Negative values are invalid.
///
/// # Errors
///
/// The graph stops and propagates an error if:
/// - The connection to the Fluvio SC fails.
/// - The consumer cannot be created (e.g. topic does not exist).
/// - An invalid `start_offset` is provided (e.g. negative).
/// - A record-level error is returned by the Fluvio cluster.
#[must_use]
pub fn fluvio_sub(
    connection: FluvioConnection,
    topic: impl Into<String>,
    partition: u32,
    start_offset: Option<i64>,
) -> Rc<dyn Stream<Burst<FluvioEvent>>> {
    let topic = topic.into();
    // Validate offset early: if negative, the stream will fail immediately on run.
    let validation_error = if let Some(n) = start_offset {
        if n < 0 {
            Some(anyhow::anyhow!(
                "start_offset must be non-negative, got {n}"
            ))
        } else {
            None
        }
    } else {
        None
    };

    produce_async(
        move |_ctx: RunParams| async move {
            // Check validation result first.
            if let Some(err) = validation_error {
                return Err(err);
            }

            let cluster_config = FluvioClusterConfig::new(&connection.endpoint);
            let client = Fluvio::connect_with_config(&cluster_config)
                .await
                .map_err(|e| anyhow::anyhow!("fluvio connect failed: {e}"))?;

            let offset = match start_offset {
                None => Offset::beginning(),
                Some(n) => Offset::absolute(n)
                    .map_err(|e| anyhow::anyhow!("invalid fluvio offset {n}: {e}"))?,
            };

            let consumer_config = ConsumerConfigExt::builder()
                .topic(topic)
                .partition(partition)
                .offset_start(offset)
                .build()
                .map_err(|e| anyhow::anyhow!("fluvio consumer config failed: {e}"))?;

            let mut stream = client
                .consumer_with_config(consumer_config)
                .await
                .map_err(|e| anyhow::anyhow!("fluvio consumer create failed: {e}"))?;

            Ok(async_stream::stream! {
                while let Some(result) = stream.next().await {
                    match result {
                        Ok(record) => {
                            let event = FluvioEvent {
                                key: record.key().map(|k| k.to_vec()),
                                value: record.value().to_vec(),
                                offset: record.offset,
                            };
                            yield Ok((NanoTime::now(), event));
                        }
                        Err(e) => {
                            yield Err(anyhow::anyhow!("fluvio record error: {e:?}"));
                            break;
                        }
                    }
                }
            })
        },
        None,
    )
}