6034126612

Committed 31 Aug 2023 07:03AM UTC coverage: 77.119%. First build

Build # 6034126612

Build Type

Pull #1945

github

Committed by

Jesse-Bakker

Commit Message

Write dozer.lock

Pull Request Pull Request #1945: Write dozer.lock

Run Details

86 of 86 new or added lines in 9 files covered. (100.0%)

49169 of 63757 relevant lines covered (77.12%)

69984.15 hits per line

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

52.08

/dozer-cli/src/simple/build/mod.rs

use dozer_api::generator::protoc::generator::ProtoGenerator;
use dozer_cache::dozer_log::{
    home_dir::{BuildId, HomeDir},
    replication::create_data_storage,
    storage::Storage,
};
use dozer_types::{log::info, models::app_config::DataStorage};
use futures::future::try_join_all;

use crate::errors::BuildError;

mod contract;

pub use contract::{Contract, PipelineContract};

pub async fn build(
    home_dir: &HomeDir,
    contract: &Contract,
    existing_contract: Option<&Contract>,
    storage_config: &DataStorage,
) -> Result<(), BuildError> {
    if let Some(build_id) =
        new_build_id(home_dir, contract, existing_contract, storage_config).await?
    {
        let build_name = build_id.name().to_string();
        build_endpoint_protos(home_dir, build_id, contract)?;
        info!("Created new build {build_name}");
    } else {
        info!("Building not needed");
    }
    Ok(())
}

async fn new_build_id(
    home_dir: &HomeDir,
    contract: &Contract,
    existing_contract: Option<&Contract>,
    storage_config: &DataStorage,
) -> Result<Option<BuildId>, BuildError> {
    let build_path = home_dir
        .find_latest_build_path()
        .map_err(|(path, error)| BuildError::FileSystem(path.into(), error))?;
    let Some(build_path) = build_path else {
        return Ok(Some(BuildId::first()));
    };

    let Some(existing_contract) = existing_contract else {
        return Ok(Some(build_path.id.next()));
    };

    let mut futures = vec![];
    for endpoint in contract.endpoints.keys() {
        let endpoint_path = build_path.get_endpoint_path(endpoint);
        let log_dir = build_path
            .data_dir
            .join(endpoint_path.log_dir_relative_to_data_dir);
        let (storage, prefix) = create_data_storage(storage_config.clone(), log_dir.into()).await?;
        futures.push(is_empty(storage, prefix));
    }
    if !try_join_all(futures)
        .await?
        .into_iter()
        .all(std::convert::identity)
    {
        return Ok(Some(build_path.id.next()));
    }

    for (endpoint, schema) in &contract.endpoints {
        if let Some(existing_schema) = existing_contract.endpoints.get(endpoint) {
            if schema == existing_schema {
                continue;
            }
        }
        return Ok(Some(build_path.id.next()));
    }
    Ok(None)
}

async fn is_empty(storage: Box<dyn Storage>, prefix: String) -> Result<bool, BuildError> {
    let objects = storage.list_objects(prefix, None).await?;
    Ok(objects.objects.is_empty())
}

fn build_endpoint_protos(
    home_dir: &HomeDir,
    build_id: BuildId,
    contract: &Contract,
) -> Result<(), BuildError> {
    let build_path = home_dir
        .create_build_dir_all(build_id)
        .map_err(|(path, error)| BuildError::FileSystem(path.into(), error))?;

    let mut resources = Vec::new();

    let proto_folder_path = build_path.contracts_dir.as_ref();
    for (endpoint_name, schema) in &contract.endpoints {
        ProtoGenerator::generate(proto_folder_path, endpoint_name, schema)?;
        resources.push(endpoint_name.clone());
    }

    let common_resources = ProtoGenerator::copy_common(proto_folder_path)?;

    // Copy common service to be included in descriptor.
    resources.extend(common_resources);

    // Generate a descriptor based on all proto files generated within sink.
    ProtoGenerator::generate_descriptor(
        proto_folder_path,
        build_path.descriptor_path.as_ref(),
        &resources,
    )?;

    Ok(())
}

1	use dozer_api::generator::protoc::generator::ProtoGenerator;
2	use dozer_cache::dozer_log::{
3	home_dir::{BuildId, HomeDir},
4	replication::create_data_storage,
5	storage::Storage,
6	};
7	use dozer_types::{log::info, models::app_config::DataStorage};
8	use futures::future::try_join_all;
9
10	use crate::errors::BuildError;
11
12	mod contract;
13
14	pub use contract::{Contract, PipelineContract};
15
16	pub async fn build(	30✔
17	home_dir: &HomeDir,	30✔
18	contract: &Contract,	30✔
19	existing_contract: Option<&Contract>,	30✔
20	storage_config: &DataStorage,	30✔
21	) -> Result<(), BuildError> {	30✔
22	if let Some(build_id) =	30✔
23	new_build_id(home_dir, contract, existing_contract, storage_config).await?	30✔
24	{	×
25	let build_name = build_id.name().to_string();	30✔
26	build_endpoint_protos(home_dir, build_id, contract)?;	30✔
27	info!("Created new build {build_name}");	15✔
28	} else {	×
29	info!("Building not needed");	×
30	}
31	Ok(())	30✔
32	}	30✔
33		×
34	async fn new_build_id(	30✔
35	home_dir: &HomeDir,	30✔
36	contract: &Contract,	30✔
37	existing_contract: Option<&Contract>,	30✔
38	storage_config: &DataStorage,	30✔
39	) -> Result<Option<BuildId>, BuildError> {	30✔
40	let build_path = home_dir	30✔
41	.find_latest_build_path()	30✔
42	.map_err(\|(path, error)\| BuildError::FileSystem(path.into(), error))?;	30✔
43	let Some(build_path) = build_path else {	30✔
44	return Ok(Some(BuildId::first()));	30✔
45	};	×
46		×
47	let Some(existing_contract) = existing_contract else {	×
48	return Ok(Some(build_path.id.next()));	×
49	};	×
50		×
51	let mut futures = vec![];	×
52	for endpoint in contract.endpoints.keys() {	×
53	let endpoint_path = build_path.get_endpoint_path(endpoint);	×
54	let log_dir = build_path	×
55	.data_dir	×
56	.join(endpoint_path.log_dir_relative_to_data_dir);	×
57	let (storage, prefix) = create_data_storage(storage_config.clone(), log_dir.into()).await?;	×
58	futures.push(is_empty(storage, prefix));	×
59	}
60	if !try_join_all(futures)	×
61	.await?	×
62	.into_iter()	×
63	.all(std::convert::identity)	×
64	{	×
65	return Ok(Some(build_path.id.next()));	×
66	}	×
67		×
68	for (endpoint, schema) in &contract.endpoints {	×
69	if let Some(existing_schema) = existing_contract.endpoints.get(endpoint) {	×
70	if schema == existing_schema {	×
71	continue;	×
72	}	×
73	}	×
74	return Ok(Some(build_path.id.next()));	×
75	}	×
76	Ok(None)	×
77	}	30✔
78		×
79	async fn is_empty(storage: Box<dyn Storage>, prefix: String) -> Result<bool, BuildError> {	×
80	let objects = storage.list_objects(prefix, None).await?;	×
81	Ok(objects.objects.is_empty())	×
82	}	×
83		×
84	fn build_endpoint_protos(	30✔
85	home_dir: &HomeDir,	30✔
86	build_id: BuildId,	30✔
87	contract: &Contract,	30✔
88	) -> Result<(), BuildError> {	30✔
89	let build_path = home_dir	30✔
90	.create_build_dir_all(build_id)	30✔
91	.map_err(\|(path, error)\| BuildError::FileSystem(path.into(), error))?;	30✔
92		×
93	let mut resources = Vec::new();	30✔
94		30✔
95	let proto_folder_path = build_path.contracts_dir.as_ref();	30✔
96	for (endpoint_name, schema) in &contract.endpoints {	60✔
97	ProtoGenerator::generate(proto_folder_path, endpoint_name, schema)?;	30✔
98	resources.push(endpoint_name.clone());	30✔
99	}	×
100		×
101	let common_resources = ProtoGenerator::copy_common(proto_folder_path)?;	30✔
102		×
103	// Copy common service to be included in descriptor.	×
104	resources.extend(common_resources);	30✔
105		30✔
106	// Generate a descriptor based on all proto files generated within sink.	30✔
107	ProtoGenerator::generate_descriptor(	30✔
108	proto_folder_path,	30✔
109	build_path.descriptor_path.as_ref(),	30✔
110	&resources,	30✔
111	)?;	30✔
112
113	Ok(())	30✔
114	}	30✔

getdozer / dozer / 6034126612

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