5954012408

Committed 23 Aug 2023 04:32PM UTC coverage: 75.86% (-0.2%) from 76.088%

Build # 5954012408

Build Type

push

github

Committed by

web-flow

Commit Message

chore: Move ContractService implementation to Contract (#1899)

* chore: Split `build.rs` to several files

* chore: Remove `serde` from `dozer-cli/Cargo.toml`

* chore: Move `ContractService` implementation to `Contract`

Run Details

461 of 461 new or added lines in 8 files covered. (100.0%)

46996 of 61951 relevant lines covered (75.86%)

73804.64 hits per line

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

64.0

/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;

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

async fn needs_build(
    home_dir: &HomeDir,
    contract: &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 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(|is_empty| is_empty)
    {
        return Ok(Some(build_path.id.next()));
    }

    let existing_contract = Contract::deserialize(&build_path)?;
    for (endpoint, schema) in &contract.endpoints {
        if let Some(existing_schema) = existing_contract.endpoints.get(endpoint) {
            if schema == existing_schema {
                continue;
            }
        } else {
            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 create_build(
    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,
    )?;

    contract.serialize(&build_path)?;

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

getdozer / dozer / 5954012408

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