switchback_protocols/
kafka.rs

1//! Kafka protocol implementation.
2
3use switchback_codec_pb::canardleteer::switchback::protocol::kafka::v1alpha1::{
4    KafkaChannelMeta, KafkaContractMeta, KafkaMessageMeta, KafkaOperationMeta, KafkaPayload,
5};
6use switchback_codec_pb::canardleteer::switchback::protocol::kafka::v1alpha1::__buffa::oneof::kafka_payload::Kind;
7use switchback_traits::ProtocolAttachment;
8
9use crate::traits::Protocol;
10use crate::wire::encode_message;
11
12/// Built-in Kafka protocol (`"kafka"`).
13#[derive(Clone, Copy, Debug, Default)]
14pub struct KafkaProtocol;
15
16impl Protocol for KafkaProtocol {
17    fn id(&self) -> &'static str {
18        "kafka"
19    }
20}
21
22impl KafkaProtocol {
23    /// Attach contract-level Kafka metadata.
24    pub fn attach_contract(&self, meta: &KafkaContractMeta) -> ProtocolAttachment {
25        attachment_from_payload(KafkaPayload {
26            kind: Some(Kind::Contract(Box::new(meta.clone()))),
27            ..Default::default()
28        })
29    }
30
31    /// Attach channel-level Kafka metadata.
32    pub fn attach_channel(&self, meta: &KafkaChannelMeta) -> ProtocolAttachment {
33        attachment_from_payload(KafkaPayload {
34            kind: Some(Kind::Channel(Box::new(meta.clone()))),
35            ..Default::default()
36        })
37    }
38
39    /// Attach operation-level Kafka metadata.
40    pub fn attach_operation(&self, meta: &KafkaOperationMeta) -> ProtocolAttachment {
41        attachment_from_payload(KafkaPayload {
42            kind: Some(Kind::Operation(Box::new(meta.clone()))),
43            ..Default::default()
44        })
45    }
46
47    /// Attach message-level Kafka metadata.
48    pub fn attach_message(&self, meta: &KafkaMessageMeta) -> ProtocolAttachment {
49        attachment_from_payload(KafkaPayload {
50            kind: Some(Kind::Message(Box::new(meta.clone()))),
51            ..Default::default()
52        })
53    }
54}
55
56fn attachment_from_payload(payload: KafkaPayload) -> ProtocolAttachment {
57    let protocol = KafkaProtocol;
58    ProtocolAttachment {
59        protocol_id: protocol.id().to_string(),
60        payload: encode_message(&payload),
61    }
62}
switchback_protocols/kafka.rs

switchback_protocols/
kafka.rs
