4763381855

Build Type

github

Committed by GitHub

Commit Message

Merge 50bf72be2 into c58df4a0b

Pull Request Pull Request #1461: feat: Make secondary index configurable

Run Details

135 of 135 new or added lines in 6 files covered. (100.0%)

34877 of 44466 relevant lines covered (78.44%)

11367.99 hits per line

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

49.66

/dozer-api/src/grpc/typed/helper.rs

use crate::generator::protoc::generator::{
    CountResponseDesc, EventDesc, QueryResponseDesc, RecordDesc, RecordWithIdDesc,
    TokenResponseDesc,
};
use crate::grpc::types_helper::map_record;
use dozer_cache::cache::CacheRecord;
use dozer_types::grpc_types::types as GrpcTypes;
use prost_reflect::{DynamicMessage, ReflectMessage, SetFieldError, Value};

use super::TypedResponse;

pub fn on_event_to_typed_response(
    op: GrpcTypes::Operation,
    event_desc: EventDesc,
) -> Result<TypedResponse, SetFieldError> {
    let mut event = DynamicMessage::new(event_desc.message);
    event.try_set_field(
        &event_desc.typ_field,
        prost_reflect::Value::EnumNumber(op.typ),
    )?;
    if let Some(old) = op.old {
        event.try_set_field(
            &event_desc.old_field,
            prost_reflect::Value::Message(internal_record_to_pb(old, &event_desc.record_desc)?),
        )?;
    }

    event.try_set_field(
        &event_desc.new_field,
        prost_reflect::Value::Message(internal_record_to_pb(
            op.new.unwrap(),
            &event_desc.record_desc,
        )?),
    )?;

    if let Some(new_id) = op.new_id {
        event.try_set_field(&event_desc.new_id_field, prost_reflect::Value::U64(new_id))?;
    }

    Ok(TypedResponse::new(event))
}

fn internal_record_to_pb(
    record: GrpcTypes::Record,
    record_desc: &RecordDesc,
) -> Result<DynamicMessage, SetFieldError> {
    let mut msg = DynamicMessage::new(record_desc.message.clone());

    // `record_desc` has more fields than `record.values` because it also contains the version field.
    // Here `zip` handles the case.
    for (field, value) in record_desc.message.fields().zip(record.values.into_iter()) {
        if let Some(v) = interval_value_to_pb(value, record_desc) {
            msg.try_set_field(&field, v)?;
        }
    }

    msg.try_set_field(
        &record_desc.version_field,
        prost_reflect::Value::U32(record.version),
    )?;

    Ok(msg)
}

fn interval_value_to_pb(
    value: GrpcTypes::Value,
    descriptor: &RecordDesc,
) -> Option<prost_reflect::Value> {
    value.value.map(|value| match value {
        GrpcTypes::value::Value::UintValue(n) => Value::U64(n),
        GrpcTypes::value::Value::IntValue(n) => Value::I64(n),
        GrpcTypes::value::Value::FloatValue(n) => Value::F64(n),
        GrpcTypes::value::Value::BoolValue(n) => Value::Bool(n),
        GrpcTypes::value::Value::StringValue(n) => Value::String(n),
        GrpcTypes::value::Value::BytesValue(n) => {
            Value::Bytes(prost_reflect::bytes::Bytes::from(n))
        }
        GrpcTypes::value::Value::Uint128Value(n) | GrpcTypes::value::Value::Int128Value(n) => {
            Value::String(n)
        }
        GrpcTypes::value::Value::PointValue(p) => {
            let point_type_desc = descriptor.point_field.message.clone();
            let x_field_desc = &descriptor.point_field.x;
            let y_field_desc = &descriptor.point_field.y;
            let mut point = DynamicMessage::new(point_type_desc);
            point.set_field(x_field_desc, prost_reflect::Value::F64(p.x));
            point.set_field(y_field_desc, prost_reflect::Value::F64(p.y));
            Value::Message(point)
        }
        GrpcTypes::value::Value::DurationValue(d) => {
            let duration_type_desc = descriptor.duration_field.message.clone();
            let value_field_desc = &descriptor.duration_field.value;
            let time_unit_field_desc = &descriptor.duration_field.time_unit;
            let mut duration = DynamicMessage::new(duration_type_desc);
            duration.set_field(value_field_desc, prost_reflect::Value::String(d.value));
            duration.set_field(
                time_unit_field_desc,
                prost_reflect::Value::String(d.time_unit),
            );
            Value::Message(duration)
        }
        GrpcTypes::value::Value::DecimalValue(d) => {
            let decimal_type_desc = descriptor.decimal_field.message.clone();
            let flags_field_desc = &descriptor.decimal_field.flags;
            let lo_field_desc = &descriptor.decimal_field.lo;
            let mid_field_desc = &descriptor.decimal_field.mid;
            let hi_field_desc = &descriptor.decimal_field.hi;
            let mut decimal = DynamicMessage::new(decimal_type_desc);
            decimal.set_field(flags_field_desc, prost_reflect::Value::U32(d.flags));
            decimal.set_field(lo_field_desc, prost_reflect::Value::U32(d.lo));
            decimal.set_field(mid_field_desc, prost_reflect::Value::U32(d.mid));
            decimal.set_field(hi_field_desc, prost_reflect::Value::U32(d.hi));
            Value::Message(decimal)
        }
        GrpcTypes::value::Value::TimestampValue(ts) => Value::Message(ts.transcode_to_dynamic()),
        GrpcTypes::value::Value::DateValue(d) => Value::String(d),
    })
}

fn internal_record_with_id_to_pb(
    record_with_id: CacheRecord,
    record_with_id_desc: &RecordWithIdDesc,
) -> Result<DynamicMessage, SetFieldError> {
    let mut msg = DynamicMessage::new(record_with_id_desc.message.clone());

    let record_with_id = map_record(record_with_id);

    let record = internal_record_to_pb(
        record_with_id.record.expect("Record is not optional"),
        &record_with_id_desc.record_desc,
    )?;
    msg.try_set_field(
        &record_with_id_desc.record_field,
        prost_reflect::Value::Message(record),
    )?;

    let id = prost_reflect::Value::U64(record_with_id.id as _);
    msg.try_set_field(&record_with_id_desc.id_field, id)?;

    Ok(msg)
}

pub fn count_response_to_typed_response(
    count: usize,
    response_desc: CountResponseDesc,
) -> Result<TypedResponse, SetFieldError> {
    let mut msg = DynamicMessage::new(response_desc.message);
    msg.try_set_field(
        &response_desc.count_field,
        prost_reflect::Value::U64(count as _),
    )?;

    Ok(TypedResponse::new(msg))
}

pub fn query_response_to_typed_response(
    records: Vec<CacheRecord>,
    response_desc: QueryResponseDesc,
) -> Result<TypedResponse, SetFieldError> {
    let mut msg = DynamicMessage::new(response_desc.message);

    let data: Result<Vec<prost_reflect::Value>, SetFieldError> = records
        .into_iter()
        .map(|record_with_id| {
            let record_with_id =
                internal_record_with_id_to_pb(record_with_id, &response_desc.record_with_id_desc)?;
            Ok(prost_reflect::Value::Message(record_with_id))
        })
        .collect();
    msg.try_set_field(
        &response_desc.records_field,
        prost_reflect::Value::List(data?),
    )?;
    Ok(TypedResponse::new(msg))
}

pub fn token_response(token: String, response_desc: TokenResponseDesc) -> TypedResponse {
    let mut msg = DynamicMessage::new(response_desc.message);
    msg.set_field(
        &response_desc.token_field,
        prost_reflect::Value::String(token),
    );
    TypedResponse::new(msg)
}

1	use crate::generator::protoc::generator::{
2	CountResponseDesc, EventDesc, QueryResponseDesc, RecordDesc, RecordWithIdDesc,
3	TokenResponseDesc,
4	};
5	use crate::grpc::types_helper::map_record;
6	use dozer_cache::cache::CacheRecord;
7	use dozer_types::grpc_types::types as GrpcTypes;
8	use prost_reflect::{DynamicMessage, ReflectMessage, SetFieldError, Value};
9
10	use super::TypedResponse;
11
12	pub fn on_event_to_typed_response(	×
13	op: GrpcTypes::Operation,	×
14	event_desc: EventDesc,	×
15	) -> Result<TypedResponse, SetFieldError> {	×
16	let mut event = DynamicMessage::new(event_desc.message);	×
17	event.try_set_field(	×
18	&event_desc.typ_field,	×
19	prost_reflect::Value::EnumNumber(op.typ),	×
20	)?;	×
21	if let Some(old) = op.old {	×
22	event.try_set_field(
23	&event_desc.old_field,	×
24	prost_reflect::Value::Message(internal_record_to_pb(old, &event_desc.record_desc)?),	×
25	)?;	×
26	}	×
27
28	event.try_set_field(
29	&event_desc.new_field,	×
30	prost_reflect::Value::Message(internal_record_to_pb(	×
31	op.new.unwrap(),	×
32	&event_desc.record_desc,	×
33	)?),	×
34	)?;	×
35
36	if let Some(new_id) = op.new_id {	×
37	event.try_set_field(&event_desc.new_id_field, prost_reflect::Value::U64(new_id))?;	×
38	}	×
39
40	Ok(TypedResponse::new(event))	×
41	}	×
42
43	fn internal_record_to_pb(	104✔
44	record: GrpcTypes::Record,	104✔
45	record_desc: &RecordDesc,	104✔
46	) -> Result<DynamicMessage, SetFieldError> {	104✔
47	let mut msg = DynamicMessage::new(record_desc.message.clone());	104✔
48
49	// `record_desc` has more fields than `record.values` because it also contains the version field.
50	// Here `zip` handles the case.
51	for (field, value) in record_desc.message.fields().zip(record.values.into_iter()) {	520✔
52	if let Some(v) = interval_value_to_pb(value, record_desc) {	520✔
53	msg.try_set_field(&field, v)?;	312✔
54	}	208✔
55	}
56
57	msg.try_set_field(	104✔
58	&record_desc.version_field,	104✔
59	prost_reflect::Value::U32(record.version),	104✔
60	)?;	104✔
61
62	Ok(msg)	104✔
63	}	104✔
64
65	fn interval_value_to_pb(	520✔
66	value: GrpcTypes::Value,	520✔
67	descriptor: &RecordDesc,	520✔
68	) -> Option<prost_reflect::Value> {	520✔
69	value.value.map(\|value\| match value {	520✔
70	GrpcTypes::value::Value::UintValue(n) => Value::U64(n),	208✔
71	GrpcTypes::value::Value::IntValue(n) => Value::I64(n),	×
72	GrpcTypes::value::Value::FloatValue(n) => Value::F64(n),	×
73	GrpcTypes::value::Value::BoolValue(n) => Value::Bool(n),	×
74	GrpcTypes::value::Value::StringValue(n) => Value::String(n),	104✔
75	GrpcTypes::value::Value::BytesValue(n) => {	×
76	Value::Bytes(prost_reflect::bytes::Bytes::from(n))	×
77	}
78	GrpcTypes::value::Value::Uint128Value(n) \| GrpcTypes::value::Value::Int128Value(n) => {	×
79	Value::String(n)	×
80	}
81	GrpcTypes::value::Value::PointValue(p) => {	×
82	let point_type_desc = descriptor.point_field.message.clone();	×
83	let x_field_desc = &descriptor.point_field.x;	×
84	let y_field_desc = &descriptor.point_field.y;	×
85	let mut point = DynamicMessage::new(point_type_desc);	×
86	point.set_field(x_field_desc, prost_reflect::Value::F64(p.x));	×
87	point.set_field(y_field_desc, prost_reflect::Value::F64(p.y));	×
88	Value::Message(point)	×
89	}
90	GrpcTypes::value::Value::DurationValue(d) => {	×
91	let duration_type_desc = descriptor.duration_field.message.clone();	×
92	let value_field_desc = &descriptor.duration_field.value;	×
93	let time_unit_field_desc = &descriptor.duration_field.time_unit;	×
94	let mut duration = DynamicMessage::new(duration_type_desc);	×
95	duration.set_field(value_field_desc, prost_reflect::Value::String(d.value));	×
96	duration.set_field(	×
97	time_unit_field_desc,	×
98	prost_reflect::Value::String(d.time_unit),	×
99	);	×
100	Value::Message(duration)	×
101	}
102	GrpcTypes::value::Value::DecimalValue(d) => {	×
103	let decimal_type_desc = descriptor.decimal_field.message.clone();	×
104	let flags_field_desc = &descriptor.decimal_field.flags;	×
105	let lo_field_desc = &descriptor.decimal_field.lo;	×
106	let mid_field_desc = &descriptor.decimal_field.mid;	×
107	let hi_field_desc = &descriptor.decimal_field.hi;	×
108	let mut decimal = DynamicMessage::new(decimal_type_desc);	×
109	decimal.set_field(flags_field_desc, prost_reflect::Value::U32(d.flags));	×
110	decimal.set_field(lo_field_desc, prost_reflect::Value::U32(d.lo));	×
111	decimal.set_field(mid_field_desc, prost_reflect::Value::U32(d.mid));	×
112	decimal.set_field(hi_field_desc, prost_reflect::Value::U32(d.hi));	×
113	Value::Message(decimal)	×
114	}
115	GrpcTypes::value::Value::TimestampValue(ts) => Value::Message(ts.transcode_to_dynamic()),	×
116	GrpcTypes::value::Value::DateValue(d) => Value::String(d),	×
117	})	520✔
118	}	520✔
119
120	fn internal_record_with_id_to_pb(	104✔
121	record_with_id: CacheRecord,	104✔
122	record_with_id_desc: &RecordWithIdDesc,	104✔
123	) -> Result<DynamicMessage, SetFieldError> {	104✔
124	let mut msg = DynamicMessage::new(record_with_id_desc.message.clone());	104✔
125		104✔
126	let record_with_id = map_record(record_with_id);	104✔
127
128	let record = internal_record_to_pb(	104✔
129	record_with_id.record.expect("Record is not optional"),	104✔
130	&record_with_id_desc.record_desc,	104✔
131	)?;	104✔
132	msg.try_set_field(	104✔
133	&record_with_id_desc.record_field,	104✔
134	prost_reflect::Value::Message(record),	104✔
135	)?;	104✔
136
137	let id = prost_reflect::Value::U64(record_with_id.id as _);	104✔
138	msg.try_set_field(&record_with_id_desc.id_field, id)?;	104✔
139
140	Ok(msg)	104✔
141	}	104✔
142
143	pub fn count_response_to_typed_response(	3✔
144	count: usize,	3✔
145	response_desc: CountResponseDesc,	3✔
146	) -> Result<TypedResponse, SetFieldError> {	3✔
147	let mut msg = DynamicMessage::new(response_desc.message);	3✔
148	msg.try_set_field(	3✔
149	&response_desc.count_field,	3✔
150	prost_reflect::Value::U64(count as _),	3✔
151	)?;	3✔
152
153	Ok(TypedResponse::new(msg))	3✔
154	}	3✔
155
156	pub fn query_response_to_typed_response(	4✔
157	records: Vec<CacheRecord>,	4✔
158	response_desc: QueryResponseDesc,	4✔
159	) -> Result<TypedResponse, SetFieldError> {	4✔
160	let mut msg = DynamicMessage::new(response_desc.message);	4✔
161		4✔
162	let data: Result<Vec<prost_reflect::Value>, SetFieldError> = records	4✔
163	.into_iter()	4✔
164	.map(\|record_with_id\| {	104✔
165	let record_with_id =	104✔
166	internal_record_with_id_to_pb(record_with_id, &response_desc.record_with_id_desc)?;	104✔
167	Ok(prost_reflect::Value::Message(record_with_id))	104✔
168	})	104✔
169	.collect();	4✔
170	msg.try_set_field(	4✔
171	&response_desc.records_field,	4✔
172	prost_reflect::Value::List(data?),	4✔
173	)?;	×
174	Ok(TypedResponse::new(msg))	4✔
175	}	4✔
176
177	pub fn token_response(token: String, response_desc: TokenResponseDesc) -> TypedResponse {	×
178	let mut msg = DynamicMessage::new(response_desc.message);	×
179	msg.set_field(	×
180	&response_desc.token_field,	×
181	prost_reflect::Value::String(token),	×
182	);	×
183	TypedResponse::new(msg)	×
184	}	×

getdozer / dozer / 4763381855

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