4023928230

Build Type

github

Committed by GitHub

Commit Message

Merge fe141bf50 into 70bd6e0ad

Pull Request Pull Request #744: feat: Implement direct insert to cache pipeline

Run Details

243 of 243 new or added lines in 16 files covered. (100.0%)

23197 of 34900 relevant lines covered (66.47%)

45617.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/dozer-orchestrator/src/simple/direct_cache_pipeline.rs

use std::sync::Arc;

use dozer_core::dag::app::{AppPipeline, PipelineEntryPoint};
use dozer_core::dag::appsource::AppSourceId;
use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
use dozer_sql::pipeline::builder::SchemaSQLContext;
use dozer_types::models::api_endpoint::ApiEndpoint;

use crate::simple::basic_processor_factory::BasicProcessorFactory;

pub fn source_to_pipeline(
    api_endpoint: &ApiEndpoint,
) -> (AppPipeline<SchemaSQLContext>, (String, u16)) {
    let api_endpoint_name = api_endpoint.name.clone();

    let p = BasicProcessorFactory::new();

    let processor_name = "direct_sink".to_string();
    let mut pipeline = AppPipeline::new();
    pipeline.add_processor(
        Arc::new(p),
        &processor_name,
        vec![PipelineEntryPoint::new(
            AppSourceId {
                id: api_endpoint_name,
                connection: None,
            },
            DEFAULT_PORT_HANDLE,
        )],
    );

    (pipeline, (processor_name, DEFAULT_PORT_HANDLE))
}

1	use std::sync::Arc;
2
3	use dozer_core::dag::app::{AppPipeline, PipelineEntryPoint};
4	use dozer_core::dag::appsource::AppSourceId;
5	use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
6	use dozer_sql::pipeline::builder::SchemaSQLContext;
7	use dozer_types::models::api_endpoint::ApiEndpoint;
8
9	use crate::simple::basic_processor_factory::BasicProcessorFactory;
10
11	pub fn source_to_pipeline(	×
12	api_endpoint: &ApiEndpoint,	×
13	) -> (AppPipeline<SchemaSQLContext>, (String, u16)) {	×
14	let api_endpoint_name = api_endpoint.name.clone();	×
15		×
16	let p = BasicProcessorFactory::new();	×
17		×
18	let processor_name = "direct_sink".to_string();	×
19	let mut pipeline = AppPipeline::new();	×
20	pipeline.add_processor(	×
21	Arc::new(p),	×
22	&processor_name,	×
23	vec![PipelineEntryPoint::new(	×
24	AppSourceId {	×
25	id: api_endpoint_name,	×
26	connection: None,	×
27	},	×
28	DEFAULT_PORT_HANDLE,	×
29	)],	×
30	);	×
31		×
32	(pipeline, (processor_name, DEFAULT_PORT_HANDLE))	×
33	}	×

getdozer / dozer / 4023928230

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous