springql 0.18.1

SpringQL: Open-source stream processor for IoT devices and in-vehicle computers
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

// This file is part of https://github.com/SpringQL/SpringQL which is licensed under MIT OR Apache-2.0. See file LICENSE-MIT or LICENSE-APACHE for full license details.

//! A simple pipeline whose source and sink are both in-memory queue.
//!
//! Usage:
//!
//! ```bash
//! cargo run --example in_memory
//! ```

use springql::{SpringConfig, SpringPipeline, SpringSourceRowBuilder};

fn push_row_to_pipeline(pipeline: &SpringPipeline, queue_name: &str) {
    let row = SpringSourceRowBuilder::default()
        .add_column("ts", "2022-01-01 13:00:00.000000000".to_string())
        .unwrap()
        .add_column("temperature", 5.3)
        .unwrap()
        .build();

    pipeline.push(queue_name, row).unwrap()
}

fn main() {
    let pipeline = SpringPipeline::new(&SpringConfig::default()).unwrap();

    pipeline
        .command(
            "
            CREATE SOURCE STREAM source_temperature_celsius (
                ts TIMESTAMP NOT NULL ROWTIME,    
                temperature FLOAT NOT NULL
            );
            ",
        )
        .unwrap();

    pipeline
        .command(
            "
            CREATE SINK STREAM sink_temperature_fahrenheit (
                ts TIMESTAMP NOT NULL ROWTIME,    
                temperature FLOAT NOT NULL
            );
            ",
        )
        .unwrap();

    pipeline
        .command(
            "
            CREATE PUMP c_to_f AS
                INSERT INTO sink_temperature_fahrenheit (ts, temperature)
                SELECT STREAM
                    source_temperature_celsius.ts,
                    32.0 + source_temperature_celsius.temperature * 1.8
                FROM source_temperature_celsius;
            ",
        )
        .unwrap();

    pipeline
        .command(
            "
            CREATE SINK WRITER queue_temperature_fahrenheit FOR sink_temperature_fahrenheit
            TYPE IN_MEMORY_QUEUE OPTIONS (
                NAME 'q_sink'
            );
            ",
        )
        .unwrap();

    pipeline
        .command(
            "
            CREATE SOURCE READER queue_temperature_celsius FOR source_temperature_celsius
            TYPE IN_MEMORY_QUEUE OPTIONS (
                NAME 'q_src'
            );
            ",
        )
        .unwrap();

    // push rows to source
    for _ in 0..10 {
        push_row_to_pipeline(&pipeline, "q_src");
    }

    // pop rows (waits forever after all source rows are fetched)
    while let Ok(row) = pipeline.pop("q_sink") {
        let ts: String = row.get_not_null_by_index(0).unwrap();
        let temperature_fahrenheit: f32 = row.get_not_null_by_index(1).unwrap();
        eprintln!("{}\t{}", ts, temperature_fahrenheit);
    }
}