4341741407

Build Type

push

github

Committed by GitHub

Commit Message

chore: Remove `TableInfo::table_name` and `TableInfo::id`. Remove parameter of `Connector::get_tables`. (#1145)

Run Details

282 of 282 new or added lines in 43 files covered. (100.0%)

29118 of 37709 relevant lines covered (77.22%)

41938.9 hits per line

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

83.52

/dozer-api/src/grpc/shared_impl/mod.rs

use dozer_cache::cache::expression::{default_limit_for_query, QueryExpression};
use dozer_cache::cache::RecordWithId;
use dozer_cache::CacheReader;
use dozer_types::grpc_types::types::Operation;
use dozer_types::log::warn;
use dozer_types::serde_json;
use dozer_types::types::Schema;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::broadcast::Receiver;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Code, Response, Status};

use crate::api_helper::{get_records, get_records_count};
use crate::auth::Access;

mod filter;

pub fn from_error(error: impl std::error::Error) -> Status {
    Status::new(Code::Internal, error.to_string())
}

fn parse_query(
    query: Option<&str>,
    default: impl FnOnce() -> QueryExpression,
) -> Result<QueryExpression, Status> {
    match query {
        Some(query) => {
            if query.is_empty() {
                Ok(default())
            } else {
                serde_json::from_str(query).map_err(from_error)
            }
        }
        None => Ok(default()),
    }
}

pub fn count(
    reader: &CacheReader,
    endpoint_name: &str,
    query: Option<&str>,
    access: Option<Access>,
) -> Result<usize, Status> {
    let mut query = parse_query(query, QueryExpression::with_no_limit)?;
    Ok(get_records_count(
        reader,
        endpoint_name,
        &mut query,
        access,
    )?)
}

pub fn query<'a>(
    reader: &'a CacheReader,
    endpoint_name: &'a str,
    query: Option<&str>,
    access: Option<Access>,
) -> Result<(&'a Schema, Vec<RecordWithId>), Status> {
    let mut query = parse_query(query, QueryExpression::with_default_limit)?;
    if query.limit.is_none() {
        query.limit = Some(default_limit_for_query());
    }
    let (schema, records) = get_records(reader, endpoint_name, &mut query, access)?;
    Ok((schema, records))
}

pub fn on_event<T: Send + 'static>(
    reader: &CacheReader,
    endpoint_name: &str,
    filter: Option<&str>,
    mut broadcast_receiver: Option<Receiver<Operation>>,
    _access: Option<Access>,
    event_mapper: impl Fn(Operation) -> Option<T> + Send + Sync + 'static,
) -> Result<Response<ReceiverStream<T>>, Status> {
    // TODO: Use access.

    if broadcast_receiver.is_none() {
        return Err(Status::unavailable(
            "on_event is not enabled. This is currently an experimental feature. Enable it in the config.",
        ));
    }

    let filter = match filter {
        Some(filter) => {
            if filter.is_empty() {
                None
            } else {
                Some(serde_json::from_str(filter).map_err(from_error)?)
            }
        }
        None => None,
    };
    let schema = reader
        .get_schema_and_indexes_by_name(endpoint_name)
        .map_err(|_| Status::invalid_argument(endpoint_name))?
        .0
        .clone();

    let (tx, rx) = tokio::sync::mpsc::channel(1);

    tokio::spawn(async move {
        loop {
            if let Some(broadcast_receiver) = broadcast_receiver.as_mut() {
                let event = broadcast_receiver.recv().await;
                match event {
                    Ok(op) => {
                        if filter::op_satisfies_filter(&op, filter.as_ref(), &schema) {
                            if let Some(event) = event_mapper(op) {
                                if (tx.send(event).await).is_err() {
                                    // receiver dropped
                                    break;
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!("Failed to receive event from broadcast channel: {}", e);
                        if e == RecvError::Closed {
                            break;
                        }
                    }
                }
            }
        }
    });

    Ok(Response::new(ReceiverStream::new(rx)))
}

1	use dozer_cache::cache::expression::{default_limit_for_query, QueryExpression};
2	use dozer_cache::cache::RecordWithId;
3	use dozer_cache::CacheReader;
4	use dozer_types::grpc_types::types::Operation;
5	use dozer_types::log::warn;
6	use dozer_types::serde_json;
7	use dozer_types::types::Schema;
8	use tokio::sync::broadcast::error::RecvError;
9	use tokio::sync::broadcast::Receiver;
10	use tokio_stream::wrappers::ReceiverStream;
11	use tonic::{Code, Response, Status};
12
13	use crate::api_helper::{get_records, get_records_count};
14	use crate::auth::Access;
15
16	mod filter;
17
18	pub fn from_error(error: impl std::error::Error) -> Status {	×
19	Status::new(Code::Internal, error.to_string())	×
20	}	×
21
22	fn parse_query(	23✔
23	query: Option<&str>,	23✔
24	default: impl FnOnce() -> QueryExpression,	23✔
25	) -> Result<QueryExpression, Status> {	23✔
26	match query {	23✔
27	Some(query) => {	16✔
28	if query.is_empty() {	16✔
29	Ok(default())	4✔
30	} else {
31	serde_json::from_str(query).map_err(from_error)	12✔
32	}
33	}
34	None => Ok(default()),	7✔
35	}
36	}	23✔
37
38	pub fn count(	9✔
39	reader: &CacheReader,	9✔
40	endpoint_name: &str,	9✔
41	query: Option<&str>,	9✔
42	access: Option<Access>,	9✔
43	) -> Result<usize, Status> {	9✔
44	let mut query = parse_query(query, QueryExpression::with_no_limit)?;	9✔
45	Ok(get_records_count(	9✔
46	reader,	9✔
47	endpoint_name,	9✔
48	&mut query,	9✔
49	access,	9✔
50	)?)	9✔
51	}	9✔
52
53	pub fn query<'a>(	14✔
54	reader: &'a CacheReader,	14✔
55	endpoint_name: &'a str,	14✔
56	query: Option<&str>,	14✔
57	access: Option<Access>,	14✔
58	) -> Result<(&'a Schema, Vec<RecordWithId>), Status> {	14✔
59	let mut query = parse_query(query, QueryExpression::with_default_limit)?;	14✔
60	if query.limit.is_none() {	14✔
61	query.limit = Some(default_limit_for_query());	3✔
62	}	11✔
63	let (schema, records) = get_records(reader, endpoint_name, &mut query, access)?;	14✔
64	Ok((schema, records))	14✔
65	}	14✔
66
67	pub fn on_event<T: Send + 'static>(	4✔
68	reader: &CacheReader,	4✔
69	endpoint_name: &str,	4✔
70	filter: Option<&str>,	4✔
71	mut broadcast_receiver: Option<Receiver<Operation>>,	4✔
72	_access: Option<Access>,	4✔
73	event_mapper: impl Fn(Operation) -> Option<T> + Send + Sync + 'static,	4✔
74	) -> Result<Response<ReceiverStream<T>>, Status> {	4✔
75	// TODO: Use access.	4✔
76		4✔
77	if broadcast_receiver.is_none() {	4✔
78	return Err(Status::unavailable(	×
79	"on_event is not enabled. This is currently an experimental feature. Enable it in the config.",	×
80	));	×
81	}	4✔
82
83	let filter = match filter {	4✔
84	Some(filter) => {	4✔
85	if filter.is_empty() {	4✔
86	None	1✔
87	} else {
88	Some(serde_json::from_str(filter).map_err(from_error)?)	3✔
89	}
90	}
91	None => None,	×
92	};
93	let schema = reader	4✔
94	.get_schema_and_indexes_by_name(endpoint_name)	4✔
95	.map_err(\|_\| Status::invalid_argument(endpoint_name))?	4✔
96	.0
97	.clone();	4✔
98		4✔
99	let (tx, rx) = tokio::sync::mpsc::channel(1);	4✔
100		4✔
101	tokio::spawn(async move {	4✔
102	loop {
103	if let Some(broadcast_receiver) = broadcast_receiver.as_mut() {	17✔
104	let event = broadcast_receiver.recv().await;	17✔
105	match event {	13✔
106	Ok(op) => {	13✔
107	if filter::op_satisfies_filter(&op, filter.as_ref(), &schema) {	13✔
108	if let Some(event) = event_mapper(op) {	3✔
109	if (tx.send(event).await).is_err() {	3✔
110	// receiver dropped
111	break;	×
112	}	3✔
113	}	×
114	}	10✔
115	}
116	Err(e) => {	×
117	warn!("Failed to receive event from broadcast channel: {}", e);	×
118	if e == RecvError::Closed {	×
119	break;	×
120	}	×
121	}
122	}
123	}	×
124	}
125	});	4✔
126		4✔
127	Ok(Response::new(ReceiverStream::new(rx)))	4✔
128	}	4✔

getdozer / dozer / 4341741407

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