krabka_connect_postgres/
schema.rs

1use bytes::Bytes;
2use krabka_schema_serde::{RegistryClient, SchemaKind, wire::encode_protobuf};
3use prost::Message as _;
4use prost_reflect::{
5    DescriptorPool, DynamicMessage, MessageDescriptor, Value,
6    prost_types::{
7        DescriptorProto, FieldDescriptorProto, FileDescriptorProto, FileDescriptorSet,
8        field_descriptor_proto::{Label, Type},
9    },
10};
11
12use crate::{
13    PostgresConnectError,
14    model::{ColumnValue, EntityDifference, EntityKey, Operation, ScalarValue},
15};
16
17const KEY_MESSAGE_INDEX: &[i32] = &[1];
18const VALUE_MESSAGE_INDEX: &[i32] = &[2];
19const PACKAGE: &str = "krabka.connect.postgres";
20const COLUMN_VALUE: &str = "ColumnValue";
21const ENTITY_KEY: &str = "EntityKey";
22const ENTITY_DIFFERENCE: &str = "EntityDifference";
23const KEY_SUBJECT: &str = "krabka-connect-postgres-key";
24const VALUE_SUBJECT: &str = "krabka-connect-postgres-value";
25const KEY_MESSAGE_TYPE: &str = "krabka.connect.postgres.EntityKey";
26const VALUE_MESSAGE_TYPE: &str = "krabka.connect.postgres.EntityDifference";
27const PROTO_SCHEMA: &str = r#"syntax = "proto3";
28package krabka.connect.postgres;
29
30message ColumnValue {
31  string name = 1;
32  string kind = 2;
33  string string_value = 3;
34  bool bool_value = 4;
35  int64 int_value = 5;
36  bytes bytes_value = 6;
37  bool is_null = 7;
38}
39
40message EntityKey {
41  string table = 1;
42  repeated ColumnValue columns = 2;
43}
44
45message EntityDifference {
46  string table = 1;
47  string operation = 2;
48  string lsn = 3;
49  repeated ColumnValue before = 4;
50  repeated ColumnValue after = 5;
51  EntityKey key = 6;
52  int64 txid = 7;
53  int64 commit_timestamp_ms = 8;
54}
55"#;
56
57#[derive(Debug, Clone)]
58pub struct PostgresProtoEncoder {
59    key_schema_id: u32,
60    value_schema_id: u32,
61    key: MessageDescriptor,
62    value: MessageDescriptor,
63    column_value: MessageDescriptor,
64}
65
66impl PostgresProtoEncoder {
67    /// Register the connector schemas and build an encoder with the allocated IDs.
68    ///
69    /// # Errors
70    /// Returns an error when Schema Registry rejects either schema or descriptor construction fails.
71    pub async fn from_registry(registry_url: &str) -> Result<Self, PostgresConnectError> {
72        let registry = RegistryClient::new(registry_url);
73        let key_schema_id = registry
74            .register(
75                KEY_SUBJECT,
76                SchemaKind::Protobuf,
77                PROTO_SCHEMA,
78                &[],
79                Some(KEY_MESSAGE_TYPE),
80            )
81            .await
82            .map_err(registry_error)?;
83        let value_schema_id = registry
84            .register(
85                VALUE_SUBJECT,
86                SchemaKind::Protobuf,
87                PROTO_SCHEMA,
88                &[],
89                Some(VALUE_MESSAGE_TYPE),
90            )
91            .await
92            .map_err(registry_error)?;
93        Self::with_schema_ids(key_schema_id, value_schema_id)
94    }
95
96    #[cfg(test)]
97    pub(crate) fn for_test() -> Result<Self, PostgresConnectError> {
98        Self::with_schema_ids(41, 57)
99    }
100
101    /// # Errors
102    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
103    fn with_schema_ids(
104        key_schema_id: u32,
105        value_schema_id: u32,
106    ) -> Result<Self, PostgresConnectError> {
107        let pool = DescriptorPool::from_file_descriptor_set(schema_descriptor_set())
108            .map_err(convert_error)?;
109
110        Ok(Self {
111            key_schema_id,
112            value_schema_id,
113            key: message_descriptor(&pool, ENTITY_KEY)?,
114            value: message_descriptor(&pool, ENTITY_DIFFERENCE)?,
115            column_value: message_descriptor(&pool, COLUMN_VALUE)?,
116        })
117    }
118
119    /// # Errors
120    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
121    pub fn encode_key(&self, key: &EntityKey) -> Result<Bytes, PostgresConnectError> {
122        let message = self.key_to_message(key)?;
123        Ok(encode_protobuf(
124            self.key_schema_id,
125            KEY_MESSAGE_INDEX,
126            &message.encode_to_vec(),
127        ))
128    }
129
130    /// # Errors
131    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
132    pub fn encode_value(&self, value: &EntityDifference) -> Result<Bytes, PostgresConnectError> {
133        let message = self.difference_to_message(value)?;
134        Ok(encode_protobuf(
135            self.value_schema_id,
136            VALUE_MESSAGE_INDEX,
137            &message.encode_to_vec(),
138        ))
139    }
140
141    fn key_to_message(&self, key: &EntityKey) -> Result<DynamicMessage, PostgresConnectError> {
142        let mut message = DynamicMessage::new(self.key.clone());
143        set_field(&mut message, "table", Value::String(key.table.clone()))?;
144        set_field(
145            &mut message,
146            "columns",
147            Value::List(self.columns_to_values(&key.columns)?),
148        )?;
149        Ok(message)
150    }
151
152    fn difference_to_message(
153        &self,
154        difference: &EntityDifference,
155    ) -> Result<DynamicMessage, PostgresConnectError> {
156        let mut message = DynamicMessage::new(self.value.clone());
157        set_field(
158            &mut message,
159            "table",
160            Value::String(difference.table.clone()),
161        )?;
162        set_field(
163            &mut message,
164            "operation",
165            Value::String(operation_name(difference.op).to_owned()),
166        )?;
167        set_field(
168            &mut message,
169            "lsn",
170            Value::String(difference.lsn.to_string()),
171        )?;
172        set_field(
173            &mut message,
174            "before",
175            Value::List(self.columns_to_values(&difference.before)?),
176        )?;
177        set_field(
178            &mut message,
179            "after",
180            Value::List(self.columns_to_values(&difference.after)?),
181        )?;
182        set_field(
183            &mut message,
184            "key",
185            Value::Message(self.key_to_message(&difference.key)?),
186        )?;
187        if let Some(txid) = difference.txid {
188            set_field(&mut message, "txid", Value::I64(txid.0))?;
189        }
190        if let Some(commit_timestamp_ms) = difference.commit_timestamp_ms {
191            set_field(
192                &mut message,
193                "commit_timestamp_ms",
194                Value::I64(commit_timestamp_ms),
195            )?;
196        }
197        Ok(message)
198    }
199
200    fn columns_to_values(
201        &self,
202        columns: &[ColumnValue],
203    ) -> Result<Vec<Value>, PostgresConnectError> {
204        columns
205            .iter()
206            .map(|column| self.column_to_value(column).map(Value::Message))
207            .collect()
208    }
209
210    fn column_to_value(
211        &self,
212        column: &ColumnValue,
213    ) -> Result<DynamicMessage, PostgresConnectError> {
214        let mut message = DynamicMessage::new(self.column_value.clone());
215        set_field(&mut message, "name", Value::String(column.name.clone()))?;
216        set_scalar_fields(&mut message, &column.value)?;
217        Ok(message)
218    }
219}
220
221fn schema_descriptor_set() -> FileDescriptorSet {
222    FileDescriptorSet {
223        file: vec![FileDescriptorProto {
224            name: Some("krabka/connect/postgres/cdc.proto".to_owned()),
225            package: Some(PACKAGE.to_owned()),
226            syntax: Some("proto3".to_owned()),
227            message_type: vec![
228                DescriptorProto {
229                    name: Some(COLUMN_VALUE.to_owned()),
230                    field: vec![
231                        field("name", 1, Type::String),
232                        field("kind", 2, Type::String),
233                        field("string_value", 3, Type::String),
234                        field("bool_value", 4, Type::Bool),
235                        field("int_value", 5, Type::Int64),
236                        field("bytes_value", 6, Type::Bytes),
237                        field("is_null", 7, Type::Bool),
238                    ],
239                    ..DescriptorProto::default()
240                },
241                DescriptorProto {
242                    name: Some(ENTITY_KEY.to_owned()),
243                    field: vec![
244                        field("table", 1, Type::String),
245                        repeated_message_field("columns", 2, COLUMN_VALUE),
246                    ],
247                    ..DescriptorProto::default()
248                },
249                DescriptorProto {
250                    name: Some(ENTITY_DIFFERENCE.to_owned()),
251                    field: vec![
252                        field("table", 1, Type::String),
253                        field("operation", 2, Type::String),
254                        field("lsn", 3, Type::String),
255                        repeated_message_field("before", 4, COLUMN_VALUE),
256                        repeated_message_field("after", 5, COLUMN_VALUE),
257                        message_field("key", 6, ENTITY_KEY),
258                        field("txid", 7, Type::Int64),
259                        field("commit_timestamp_ms", 8, Type::Int64),
260                    ],
261                    ..DescriptorProto::default()
262                },
263            ],
264            ..FileDescriptorProto::default()
265        }],
266    }
267}
268
269fn field(name: &str, number: i32, field_type: Type) -> FieldDescriptorProto {
270    FieldDescriptorProto {
271        name: Some(name.to_owned()),
272        number: Some(number),
273        label: Some(Label::Optional as i32),
274        r#type: Some(field_type as i32),
275        ..FieldDescriptorProto::default()
276    }
277}
278
279fn message_field(name: &str, number: i32, message_name: &str) -> FieldDescriptorProto {
280    FieldDescriptorProto {
281        type_name: Some(full_type_name(message_name)),
282        ..field(name, number, Type::Message)
283    }
284}
285
286fn repeated_message_field(name: &str, number: i32, message_name: &str) -> FieldDescriptorProto {
287    FieldDescriptorProto {
288        label: Some(Label::Repeated as i32),
289        ..message_field(name, number, message_name)
290    }
291}
292
293fn full_type_name(message_name: &str) -> String {
294    format!(".{PACKAGE}.{message_name}")
295}
296
297fn message_descriptor(
298    pool: &DescriptorPool,
299    name: &str,
300) -> Result<MessageDescriptor, PostgresConnectError> {
301    pool.get_message_by_name(&format!("{PACKAGE}.{name}"))
302        .ok_or_else(|| PostgresConnectError::Convert(format!("protobuf message {name} not found")))
303}
304
305fn set_field(
306    message: &mut DynamicMessage,
307    name: &str,
308    value: Value,
309) -> Result<(), PostgresConnectError> {
310    message
311        .try_set_field_by_name(name, value)
312        .map_err(convert_error)
313}
314
315fn operation_name(operation: Operation) -> &'static str {
316    match operation {
317        Operation::Insert => "insert",
318        Operation::Update => "update",
319        Operation::Delete => "delete",
320    }
321}
322
323fn set_scalar_fields(
324    message: &mut DynamicMessage,
325    value: &ScalarValue,
326) -> Result<(), PostgresConnectError> {
327    match value {
328        ScalarValue::Null => {
329            set_field(message, "kind", Value::String("null".to_owned()))?;
330            set_field(message, "is_null", Value::Bool(true))?;
331        }
332        ScalarValue::UnchangedToast => {
333            set_field(message, "kind", Value::String("unchanged_toast".to_owned()))?;
334        }
335        ScalarValue::Bool(value) => {
336            set_field(message, "kind", Value::String("bool".to_owned()))?;
337            set_field(message, "bool_value", Value::Bool(*value))?;
338        }
339        ScalarValue::Int(value) => {
340            set_field(message, "kind", Value::String("int".to_owned()))?;
341            set_field(message, "int_value", Value::I64(*value))?;
342        }
343        ScalarValue::Float(value) => {
344            set_field(message, "kind", Value::String("float".to_owned()))?;
345            set_field(message, "string_value", Value::String(value.clone()))?;
346        }
347        ScalarValue::Text(value) => {
348            set_field(message, "kind", Value::String("text".to_owned()))?;
349            set_field(message, "string_value", Value::String(value.clone()))?;
350        }
351        ScalarValue::Bytes(value) => {
352            set_field(message, "kind", Value::String("bytes".to_owned()))?;
353            set_field(
354                message,
355                "bytes_value",
356                Value::Bytes(Bytes::copy_from_slice(value)),
357            )?;
358        }
359    }
360
361    Ok(())
362}
363
364fn convert_error(error: impl std::fmt::Display) -> PostgresConnectError {
365    PostgresConnectError::Convert(error.to_string())
366}
367
368fn registry_error(error: impl std::fmt::Display) -> PostgresConnectError {
369    PostgresConnectError::Convert(format!("schema registry: {error}"))
370}
371
372#[cfg(test)]
373mod tests {
374    use bytes::Bytes;
375    use krabka_schema_serde::wire::decode_protobuf;
376    use prost_reflect::{DescriptorPool, DynamicMessage, Value};
377    use wiremock::{
378        Mock, MockServer, ResponseTemplate,
379        matchers::{body_json, method, path},
380    };
381
382    use super::{
383        COLUMN_VALUE, ENTITY_DIFFERENCE, ENTITY_KEY, KEY_MESSAGE_TYPE, PROTO_SCHEMA,
384        PostgresProtoEncoder, VALUE_MESSAGE_TYPE, message_descriptor, schema_descriptor_set,
385    };
386    use crate::{
387        ColumnValue, EntityDifference, EntityKey, Operation, PgLsn, TableSchema,
388        ids::{RelationId, TransactionId},
389        model::{ColumnSchema, ScalarValue},
390        pgoutput::{RelationCache, RelationEvent, RowEvent, RowEventKind},
391    };
392
393    #[test]
394    fn encoder_frames_key_and_value_as_protobuf() {
395        let encoder =
396            PostgresProtoEncoder::with_schema_ids(41, 57).expect("encoder builds descriptors");
397        let diff = sample_difference();
398        let pool = DescriptorPool::from_file_descriptor_set(schema_descriptor_set())
399            .expect("descriptor pool builds");
400
401        let key = encoder.encode_key(&diff.key).expect("key encodes");
402        let value = encoder.encode_value(&diff).expect("value encodes");
403
404        let (key_id, key_index, key_body) = decode_protobuf(&key).expect("key frame decodes");
405        let key_frame = (key_id, key_index, key_body.is_empty());
406
407        let (value_id, value_index, value_body) =
408            decode_protobuf(&value).expect("value frame decodes");
409        let value_frame = (value_id, value_index, value_body.is_empty());
410
411        let key_message = DynamicMessage::decode(
412            message_descriptor(&pool, ENTITY_KEY).expect("key descriptor"),
413            key_body,
414        )
415        .expect("key body decodes");
416        let key_columns = list_field(&key_message, "columns");
417        let id_column = message_value(&key_columns[0]);
418        let key_projection = (
419            string_field(&key_message, "table"),
420            string_field(id_column, "name"),
421            string_field(id_column, "kind"),
422            i64_field(id_column, "int_value"),
423        );
424
425        let value_message = DynamicMessage::decode(
426            message_descriptor(&pool, ENTITY_DIFFERENCE).expect("value descriptor"),
427            value_body,
428        )
429        .expect("value body decodes");
430        let value_projection = (
431            string_field(&value_message, "table"),
432            string_field(&value_message, "operation"),
433            string_field(&value_message, "lsn"),
434        );
435
436        let after = list_field(&value_message, "after");
437        let name_column = message_value(&after[0]);
438        let name_projection = (
439            string_field(name_column, "name"),
440            string_field(name_column, "kind"),
441            string_field(name_column, "string_value"),
442        );
443
444        let before = list_field(&value_message, "before");
445        let null_column = message_value(&before[1]);
446        let null_projection = (
447            string_field(null_column, "name"),
448            string_field(null_column, "kind"),
449            bool_field(null_column, "is_null"),
450        );
451
452        let avatar_column = message_value(&after[1]);
453        let avatar_projection = (
454            string_field(avatar_column, "name"),
455            string_field(avatar_column, "kind"),
456            bytes_field(avatar_column, "bytes_value"),
457        );
458
459        let unchanged_toast_column = message_value(&after[2]);
460        let unchanged_projection = (
461            string_field(unchanged_toast_column, "name"),
462            string_field(unchanged_toast_column, "kind"),
463        );
464
465        assert2::assert!(key_frame == (41, vec![1], false));
466        assert2::assert!(value_frame == (57, vec![2], false));
467        assert2::assert!(
468            key_projection
469                == (
470                    "public.accounts".to_string(),
471                    "id".to_string(),
472                    "int".to_string(),
473                    42,
474                )
475        );
476        assert2::assert!(
477            value_projection
478                == (
479                    "public.accounts".to_string(),
480                    "update".to_string(),
481                    "0/2A".to_string(),
482                )
483        );
484        assert2::assert!(
485            name_projection == ("name".to_string(), "text".to_string(), "new".to_string())
486        );
487        assert2::assert!(null_projection == ("nickname".to_string(), "null".to_string(), true));
488        assert2::assert!(
489            avatar_projection
490                == (
491                    "avatar".to_string(),
492                    "bytes".to_string(),
493                    Bytes::from_static(b"abc"),
494                )
495        );
496        assert2::assert!(
497            unchanged_projection == ("details".to_string(), "unchanged_toast".to_string())
498        );
499    }
500
501    #[tokio::test]
502    async fn registry_allocates_ids_used_by_key_and_value_frames() {
503        let server = MockServer::start().await;
504        for (subject, message_type, id) in [
505            ("krabka-connect-postgres-key", KEY_MESSAGE_TYPE, 73),
506            ("krabka-connect-postgres-value", VALUE_MESSAGE_TYPE, 91),
507        ] {
508            Mock::given(method("POST"))
509                .and(path(format!("/subjects/{subject}/versions")))
510                .and(body_json(serde_json::json!({
511                    "schema": PROTO_SCHEMA,
512                    "schemaType": "PROTOBUF",
513                    "messageType": message_type,
514                })))
515                .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
516                    "id": id,
517                })))
518                .expect(1)
519                .mount(&server)
520                .await;
521        }
522
523        let encoder = PostgresProtoEncoder::from_registry(&server.uri())
524            .await
525            .expect("registry allocates both IDs");
526        let difference = sample_difference();
527        let key = encoder.encode_key(&difference.key).expect("key encodes");
528        let value = encoder.encode_value(&difference).expect("value encodes");
529
530        assert2::assert!(decode_protobuf(&key).expect("key frame").0 == 73);
531        assert2::assert!(decode_protobuf(&value).expect("value frame").0 == 91);
532    }
533
534    #[test]
535    fn decoded_int8_key_encodes_as_int_scalar_kind() {
536        let mut cache = RelationCache::default();
537        cache.apply_relation(RelationEvent {
538            relation_id: RelationId(7),
539            schema: "public".to_owned(),
540            table: "orders".to_owned(),
541            columns: vec![ColumnSchema {
542                name: "id".to_owned(),
543                type_name: "int8".to_owned(),
544                key: true,
545            }],
546        });
547        let difference = cache
548            .translate(RowEvent {
549                relation_id: RelationId(7),
550                lsn: PgLsn(0x2a),
551                commit_lsn: None,
552                txid: None,
553                commit_timestamp_ms: None,
554                kind: RowEventKind::Insert,
555                values: vec![ColumnValue {
556                    name: "col0".to_owned(),
557                    value: ScalarValue::Text("42".to_owned()),
558                }],
559            })
560            .expect("decoded row should translate");
561        assert2::assert!(
562            &difference
563                == &EntityDifference {
564                    table: "public.orders".into(),
565                    key: EntityKey {
566                        table: "public.orders".into(),
567                        columns: vec![ColumnValue {
568                            name: "id".into(),
569                            value: ScalarValue::Int(42),
570                        }],
571                    },
572                    op: Operation::Insert,
573                    before: vec![],
574                    after: vec![ColumnValue {
575                        name: "id".into(),
576                        value: ScalarValue::Int(42),
577                    }],
578                    lsn: PgLsn(0x2a),
579                    txid: None,
580                    commit_timestamp_ms: None,
581                    schema: TableSchema {
582                        schema: "public".into(),
583                        table: "orders".into(),
584                        columns: vec![ColumnSchema {
585                            name: "id".into(),
586                            type_name: "int8".into(),
587                            key: true,
588                        }],
589                    },
590                }
591        );
592
593        let encoder =
594            PostgresProtoEncoder::with_schema_ids(41, 57).expect("encoder builds descriptors");
595        let key = encoder.encode_key(&difference.key).expect("key encodes");
596        let pool = DescriptorPool::from_file_descriptor_set(schema_descriptor_set())
597            .expect("descriptor pool builds");
598        let (_, _, key_body) = decode_protobuf(&key).expect("key frame decodes");
599        let key_message = DynamicMessage::decode(
600            message_descriptor(&pool, ENTITY_KEY).expect("key descriptor"),
601            key_body,
602        )
603        .expect("key body decodes");
604        let key_columns = list_field(&key_message, "columns");
605        let id_column = message_value(&key_columns[0]);
606
607        assert2::assert!(string_field(id_column, "kind") == "int".to_string());
608        assert2::assert!(i64_field(id_column, "int_value") == 42);
609    }
610
611    #[test]
612    fn descriptor_fields_have_expected_proto3_labels() {
613        let descriptor_set = schema_descriptor_set();
614        let file = descriptor_set.file.first().expect("descriptor file");
615        let column_value = file
616            .message_type
617            .iter()
618            .find(|message| message.name.as_deref() == Some(COLUMN_VALUE))
619            .expect("column value message");
620        let entity_key = file
621            .message_type
622            .iter()
623            .find(|message| message.name.as_deref() == Some(ENTITY_KEY))
624            .expect("entity key message");
625        let entity_difference = file
626            .message_type
627            .iter()
628            .find(|message| message.name.as_deref() == Some(ENTITY_DIFFERENCE))
629            .expect("entity difference message");
630
631        assert2::assert!(
632            [
633                column_value
634                    .field
635                    .iter()
636                    .find(|field| field.name.as_deref() == Some("name"))
637                    .and_then(|field| field.label),
638                entity_key
639                    .field
640                    .iter()
641                    .find(|field| field.name.as_deref() == Some("columns"))
642                    .and_then(|field| field.label),
643                entity_difference
644                    .field
645                    .iter()
646                    .find(|field| field.name.as_deref() == Some("key"))
647                    .and_then(|field| field.label),
648            ] == [
649                Some(prost_reflect::prost_types::field_descriptor_proto::Label::Optional as i32),
650                Some(prost_reflect::prost_types::field_descriptor_proto::Label::Repeated as i32),
651                Some(prost_reflect::prost_types::field_descriptor_proto::Label::Optional as i32),
652            ]
653        );
654    }
655
656    fn sample_difference() -> EntityDifference {
657        let key = EntityKey {
658            table: "public.accounts".to_owned(),
659            columns: vec![ColumnValue {
660                name: "id".to_owned(),
661                value: ScalarValue::Int(42),
662            }],
663        };
664
665        EntityDifference {
666            table: "public.accounts".to_owned(),
667            key,
668            op: Operation::Update,
669            before: vec![
670                ColumnValue {
671                    name: "name".to_owned(),
672                    value: ScalarValue::Text("old".to_owned()),
673                },
674                ColumnValue {
675                    name: "nickname".to_owned(),
676                    value: ScalarValue::Null,
677                },
678            ],
679            after: vec![
680                ColumnValue {
681                    name: "name".to_owned(),
682                    value: ScalarValue::Text("new".to_owned()),
683                },
684                ColumnValue {
685                    name: "avatar".to_owned(),
686                    value: ScalarValue::Bytes(b"abc".to_vec()),
687                },
688                ColumnValue {
689                    name: "details".to_owned(),
690                    value: ScalarValue::UnchangedToast,
691                },
692            ],
693            lsn: PgLsn(42),
694            txid: Some(TransactionId(7)),
695            commit_timestamp_ms: Some(1_700_000_000_000),
696            schema: TableSchema {
697                schema: "public".to_owned(),
698                table: "accounts".to_owned(),
699                columns: vec![
700                    ColumnSchema {
701                        name: "id".to_owned(),
702                        type_name: "int8".to_owned(),
703                        key: true,
704                    },
705                    ColumnSchema {
706                        name: "name".to_owned(),
707                        type_name: "text".to_owned(),
708                        key: false,
709                    },
710                ],
711            },
712        }
713    }
714
715    fn string_field(message: &DynamicMessage, name: &str) -> String {
716        match message
717            .get_field_by_name(name)
718            .expect("field exists")
719            .as_ref()
720        {
721            Value::String(value) => value.clone(),
722            other => panic!("field {name} was not a string: {other:?}"),
723        }
724    }
725
726    fn bool_field(message: &DynamicMessage, name: &str) -> bool {
727        match message
728            .get_field_by_name(name)
729            .expect("field exists")
730            .as_ref()
731        {
732            Value::Bool(value) => *value,
733            other => panic!("field {name} was not a bool: {other:?}"),
734        }
735    }
736
737    fn i64_field(message: &DynamicMessage, name: &str) -> i64 {
738        match message
739            .get_field_by_name(name)
740            .expect("field exists")
741            .as_ref()
742        {
743            Value::I64(value) => *value,
744            other => panic!("field {name} was not an int64: {other:?}"),
745        }
746    }
747
748    fn bytes_field(message: &DynamicMessage, name: &str) -> Bytes {
749        match message
750            .get_field_by_name(name)
751            .expect("field exists")
752            .as_ref()
753        {
754            Value::Bytes(value) => value.clone(),
755            other => panic!("field {name} was not bytes: {other:?}"),
756        }
757    }
758
759    fn list_field(message: &DynamicMessage, name: &str) -> Vec<Value> {
760        match message
761            .get_field_by_name(name)
762            .expect("field exists")
763            .as_ref()
764        {
765            Value::List(value) => value.clone(),
766            other => panic!("field {name} was not a list: {other:?}"),
767        }
768    }
769
770    fn message_value(value: &Value) -> &DynamicMessage {
771        match value {
772            Value::Message(message) => message,
773            other => panic!("value was not a message: {other:?}"),
774        }
775    }
776}
krabka_connect_postgres/schema.rs

krabka_connect_postgres/
schema.rs
