springql-core 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.

use std::sync::Arc;

use crate::{
    api::error::Result,
    pipeline::{
        Pipeline, PipelineVersion, PumpModel, SinkWriterModel, SourceReaderModel, StreamModel,
    },
    stream_engine::command::AlterPipelineCommand,
};

/// Executor of SQL.
///
/// All methods (recursive) are called from main thread.
#[derive(Debug)]
pub struct SqlExecutor {
    pipeline: Pipeline,
}

impl Default for SqlExecutor {
    fn default() -> Self {
        Self {
            pipeline: Pipeline::new(PipelineVersion::new()),
        }
    }
}

impl SqlExecutor {
    pub fn current_pipeline(&self) -> &Pipeline {
        &self.pipeline
    }

    pub fn alter_pipeline(&mut self, command: AlterPipelineCommand) -> Result<Pipeline> {
        let new_pipeline = Self::new_pipeline(self.pipeline.clone(), command)?;
        self.pipeline = new_pipeline;
        Ok(self.pipeline.clone())
    }

    fn new_pipeline(pipeline: Pipeline, command: AlterPipelineCommand) -> Result<Pipeline> {
        match command {
            AlterPipelineCommand::CreateSourceStream(source_stream) => {
                Self::create_source_stream(pipeline, source_stream)
            }
            AlterPipelineCommand::CreateSourceReader(source_reader) => {
                Self::create_source_reader(pipeline, source_reader)
            }
            AlterPipelineCommand::CreateStream(stream) => Self::create_stream(pipeline, stream),
            AlterPipelineCommand::CreateSinkStream(sink_stream) => {
                Self::create_sink_stream(pipeline, sink_stream)
            }
            AlterPipelineCommand::CreateSinkWriter(sink_writer) => {
                Self::create_sink_writer(pipeline, sink_writer)
            }
            AlterPipelineCommand::CreatePump(pump) => Self::create_pump(pipeline, *pump),
        }
    }

    fn create_source_stream(
        mut pipeline: Pipeline,
        source_stream: StreamModel,
    ) -> Result<Pipeline> {
        pipeline.add_stream(Arc::new(source_stream))?;
        Ok(pipeline)
    }
    fn create_source_reader(
        mut pipeline: Pipeline,
        source_reader: SourceReaderModel,
    ) -> Result<Pipeline> {
        pipeline.add_source_reader(source_reader)?;
        Ok(pipeline)
    }

    fn create_stream(mut pipeline: Pipeline, stream: StreamModel) -> Result<Pipeline> {
        pipeline.add_stream(Arc::new(stream))?;
        Ok(pipeline)
    }

    fn create_sink_stream(mut pipeline: Pipeline, sink_stream: StreamModel) -> Result<Pipeline> {
        pipeline.add_stream(Arc::new(sink_stream))?;
        Ok(pipeline)
    }
    fn create_sink_writer(
        mut pipeline: Pipeline,
        sink_writer: SinkWriterModel,
    ) -> Result<Pipeline> {
        pipeline.add_sink_writer(sink_writer)?;
        Ok(pipeline)
    }

    fn create_pump(mut pipeline: Pipeline, pump: PumpModel) -> Result<Pipeline> {
        pipeline.add_pump(pump)?;
        Ok(pipeline)
    }
}