tinyflow-api 0.1.1

Core traits, types, and abstractions for Tinyflow streaming 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

use crate::checkpoint::CheckpointContext;
use crate::context::RuntimeContext;
use crate::error::StreamResult;
use async_trait::async_trait;
use std::collections::HashMap;

#[async_trait]
pub trait LifecycleAware {
    async fn open(&mut self, _ctx: &mut RuntimeContext) -> StreamResult<()> {
        Ok(())
    }

    async fn close(&mut self, _ctx: &mut RuntimeContext) -> StreamResult<()> {
        Ok(())
    }

    async fn on_checkpoint(
        &mut self,
        _runtime: &mut RuntimeContext,
        _ckpt: &CheckpointContext,
    ) -> StreamResult<()> {
        Ok(())
    }

    async fn on_checkpoint_end(
        &mut self,
        _runtime: &mut RuntimeContext,
        _ckpt: &CheckpointContext,
    ) -> StreamResult<()> {
        Ok(())
    }

    fn report_metrics(&self) -> HashMap<String, f64> {
        HashMap::new()
    }
}

#[macro_export]
macro_rules! lifecycle {
    ($t:ty) => {
        impl $crate::functions::LifecycleAware for $t {}
    };
}

#[async_trait]
pub trait MapFunction<T, O>: LifecycleAware {
    async fn map(&mut self, input: T, runtime: &mut RuntimeContext) -> StreamResult<O>;
}

#[async_trait]
pub trait FilterFunction<T>: LifecycleAware {
    async fn filter(&mut self, input: &T, runtime: &mut RuntimeContext) -> StreamResult<bool>;
}

#[async_trait]
pub trait FlatMapFunction<T, O>: LifecycleAware {
    async fn flat_map(&mut self, input: T, runtime: &mut RuntimeContext) -> StreamResult<Vec<O>>;
}

#[async_trait]
pub trait SinkFunction<T>: LifecycleAware {
    async fn accept(&mut self, input: T, runtime: &mut RuntimeContext) -> StreamResult<()>;
}

#[async_trait]
pub trait SourceFunction<T: Send>: LifecycleAware {
    async fn poll_next(&mut self) -> StreamResult<Option<T>>;
}