krabka_sdk_conformance/
protocol.rs

1//! JSON-lines protocol shared by the conformance harness and SDK adapters.
2
3use serde::{Deserialize, Serialize};
4
5/// Major version for the v1 SDK conformance contract.
6pub const CONTRACT_MAJOR: u16 = 1;
7
8/// Minor version for adapters that only implement the original v1.0 contract.
9pub const CONTRACT_MINOR_V1_0: u16 = 0;
10
11/// Minor version for adapters that implement queue RPC shapes.
12pub const CONTRACT_MINOR_QUEUE_RPC: u16 = 1;
13
14/// A command sent from the conformance harness to a language adapter.
15#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
16#[serde(tag = "cmd", rename_all = "snake_case")]
17pub enum Command {
18    /// Negotiate the contract major and adapter language.
19    Hello,
20    /// Configure the SDK endpoint and optional bearer token.
21    Configure {
22        /// Gateway endpoint URL.
23        endpoint: String,
24        /// Optional bearer token to forward through the SDK.
25        bearer: Option<String>,
26    },
27    /// Publish a raw record.
28    Publish {
29        /// Topic name.
30        topic: String,
31        /// Base64-encoded record value.
32        value_b64: String,
33        /// Binary headers in publish order.
34        headers: Vec<Header>,
35    },
36    /// Publish a `CloudEvent` in binary mode.
37    PublishEvent {
38        /// Topic name.
39        topic: String,
40        /// `CloudEvent` payload and attributes.
41        event: CloudEvent,
42    },
43    /// Open or replace a subscription.
44    Subscribe {
45        /// Topics to subscribe to.
46        topics: Vec<String>,
47        /// Consumer group id.
48        group: String,
49        /// Optional structured equality filter.
50        filter: Option<Filter>,
51    },
52    /// Pull the next message from the active subscription.
53    NextMessage {
54        /// Maximum wait time in milliseconds.
55        timeout_ms: u64,
56    },
57    /// Queue acquire stub command.
58    QueueAcquire {
59        /// Topic/queue name.
60        topic: String,
61        /// Worker group id.
62        group: String,
63        /// Maximum number of messages.
64        max: u32,
65        /// Lock duration in milliseconds.
66        lock_duration_ms: u64,
67        /// Optional session id returned by a prior acquire.
68        #[serde(default)]
69        session_id: String,
70    },
71    /// Queue ack stub command kept in the v1.0 stub-vector group.
72    QueueAck {
73        /// Message id to ack.
74        message_id: String,
75    },
76    /// Queue acknowledge command for the v1.1 queue RPC contract group.
77    QueueAcknowledge {
78        /// Session id returned by queue acquire.
79        session_id: String,
80        /// Entries to acknowledge.
81        entries: Vec<QueueAckEntry>,
82    },
83    /// Queue renew command for the v1.1 queue RPC contract group.
84    QueueRenew {
85        /// Session id returned by queue acquire.
86        session_id: String,
87        /// Entries to renew.
88        entries: Vec<QueueRenewEntry>,
89    },
90    /// Database connect stub command.
91    DbConnect {
92        /// Database name.
93        name: String,
94    },
95    /// Auth sign-in placeholder command.
96    AuthSignIn {
97        /// Username.
98        username: String,
99        /// Password.
100        password: String,
101    },
102    /// Blob put stub command.
103    BlobPut {
104        /// Blob key.
105        key: String,
106        /// Base64-encoded blob body.
107        value_b64: String,
108    },
109    /// Blob get stub command.
110    BlobGet {
111        /// Blob key.
112        key: String,
113    },
114}
115
116/// A response sent by an adapter for a single [`Command`].
117#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
118#[serde(rename_all = "snake_case")]
119pub enum Response {
120    /// Contract negotiation response.
121    Hello {
122        /// Adapter contract major.
123        contract_major: u16,
124        /// Adapter contract minor.
125        contract_minor: u16,
126        /// Adapter language name.
127        language: String,
128    },
129    /// Successful command response.
130    Ok(serde_json::Value),
131    /// Message delivered by a subscription.
132    Message(Message),
133    /// Adapter or SDK error.
134    Error(AdapterError),
135}
136
137/// Wire error taxonomy shared across language SDKs.
138#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
139#[serde(rename_all = "snake_case")]
140pub enum ErrorKind {
141    /// Transport or endpoint reachability failure.
142    Transport,
143    /// Authentication failed or credentials were absent.
144    Unauthenticated,
145    /// Caller supplied an invalid argument.
146    InvalidArgument,
147    /// Target resource was not found.
148    NotFound,
149    /// Server-side failure outside the narrower classes.
150    ServerError,
151    /// Module is intentionally gated on later work.
152    Unimplemented,
153}
154
155/// Error body returned by adapters.
156#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
157pub struct AdapterError {
158    /// Taxonomy kind.
159    pub kind: ErrorKind,
160    /// Optional SDK module name.
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub module: Option<String>,
163    /// Optional plan/spec slug gating an unimplemented module.
164    #[serde(skip_serializing_if = "Option::is_none")]
165    pub gated_on: Option<String>,
166    /// Optional diagnostic message.
167    #[serde(skip_serializing_if = "Option::is_none")]
168    pub message: Option<String>,
169}
170
171impl AdapterError {
172    /// Build a pinned unimplemented-module error.
173    #[must_use]
174    pub fn unimplemented(module: &str, gated_on: &str) -> Self {
175        Self {
176            kind: ErrorKind::Unimplemented,
177            module: Some(module.to_string()),
178            gated_on: Some(gated_on.to_string()),
179            message: None,
180        }
181    }
182
183    /// Build an error with only a kind and message.
184    #[must_use]
185    pub fn with_message(kind: ErrorKind, message: impl Into<String>) -> Self {
186        Self {
187            kind,
188            module: None,
189            gated_on: None,
190            message: Some(message.into()),
191        }
192    }
193}
194
195/// Binary record header.
196#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
197pub struct Header {
198    /// Header name.
199    pub name: String,
200    /// Base64-encoded header value, or null for a Kafka null header value.
201    pub value_b64: Option<String>,
202}
203
204/// Message response body.
205#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
206pub struct Message {
207    /// Topic name.
208    pub topic: String,
209    /// Partition index.
210    pub partition: i32,
211    /// Record offset.
212    pub offset: i64,
213    /// Base64-encoded record value.
214    pub value_b64: String,
215    /// Binary headers in delivery order.
216    pub headers: Vec<Header>,
217}
218
219/// Queue acknowledgement type.
220#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
221#[serde(rename_all = "snake_case")]
222pub enum QueueAckType {
223    /// Accept and remove the queue message.
224    Accept,
225    /// Release the queue message for redelivery.
226    Release,
227    /// Reject the queue message.
228    Reject,
229}
230
231/// Queue acknowledgement entry.
232#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
233pub struct QueueAckEntry {
234    /// Message id returned by queue acquire.
235    pub message_id: String,
236    /// Ack verdict.
237    pub ack_type: QueueAckType,
238}
239
240/// Queue renew entry.
241#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
242pub struct QueueRenewEntry {
243    /// Message id returned by queue acquire.
244    pub message_id: String,
245}
246
247/// Minimal `CloudEvents` v1 binary-mode payload.
248#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
249pub struct CloudEvent {
250    /// Event id.
251    pub id: String,
252    /// Event source.
253    pub source: String,
254    /// Event type.
255    #[serde(rename = "type")]
256    pub type_: String,
257    /// `CloudEvents` specversion.
258    pub specversion: String,
259    /// Optional data content type.
260    #[serde(skip_serializing_if = "Option::is_none")]
261    pub datacontenttype: Option<String>,
262    /// Base64-encoded event data.
263    pub data_b64: String,
264}
265
266/// Structured subscription filter.
267#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
268pub struct Filter {
269    /// JSON field path. v1 mock vectors support `$.field`.
270    pub path: String,
271    /// Predicate operation.
272    pub op: FilterOp,
273    /// Expected scalar JSON value.
274    pub value: serde_json::Value,
275}
276
277/// Supported filter operation.
278#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
279#[serde(rename_all = "snake_case")]
280pub enum FilterOp {
281    /// Equality comparison.
282    Equals,
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288
289    #[test]
290    fn protocol_round_trips() {
291        let cmd = Command::Publish {
292            topic: "t".into(),
293            value_b64: "aGk=".into(),
294            headers: vec![],
295        };
296        let line = serde_json::to_string(&cmd).unwrap();
297        assert_eq!(serde_json::from_str::<Command>(&line).unwrap(), cmd);
298
299        let err: Response = serde_json::from_str(
300            r#"{"error":{"kind":"unimplemented","module":"queues","gated_on":"gateway-sharegroup-rpc"}}"#,
301        )
302        .unwrap();
303        assert2::assert!(let Response::Error(e) = err);
304        assert_eq!(e.kind, ErrorKind::Unimplemented);
305        assert_eq!(e.gated_on.as_deref(), Some("gateway-sharegroup-rpc"));
306    }
307}
krabka_sdk_conformance/protocol.rs

krabka_sdk_conformance/
protocol.rs
