netget/server/pulsar/
wire.rs

1//! The Pulsar binary protocol for both roles: the protobuf commands NetGet speaks (a subset of
2//! `PulsarApi.proto`, field numbers and proto2 `required` labels as upstream defines them, so
3//! a zero-valued required field is still written), frame encoding, the payload section of
4//! SEND and MESSAGE frames (magic, CRC32C, metadata), and batch entries.
5use anyhow::{bail, ensure, Context, Result};
6use prost::Message;
7use std::time::Duration;
8use tokio::io::{AsyncRead, AsyncReadExt};
9
10/// A frame (all of it, after the 4-byte total size) at most; Pulsar's own default
11/// `maxMessageSize` is 5 MiB, and NetGet advertises this number in CONNECTED.
12pub const MAX_FRAME: usize = 5 * 1024 * 1024;
13/// A command's protobuf at most.
14pub const MAX_COMMAND: usize = 64 * 1024;
15/// Messages one batch may carry.
16pub const MAX_BATCH: usize = 1000;
17pub const IO_TIMEOUT: Duration = Duration::from_secs(30);
18/// The protocol version NetGet speaks (v19: transaction-coordinator connect; NetGet uses none
19/// of the newer features but nothing in them changes the framing).
20pub const PROTOCOL_VERSION: i32 = 19;
21const MAGIC_CRC32C: u16 = 0x0e01;
22
23pub mod command_type {
24    pub const CONNECT: i32 = 2;
25    pub const CONNECTED: i32 = 3;
26    pub const SUBSCRIBE: i32 = 4;
27    pub const PRODUCER: i32 = 5;
28    pub const SEND: i32 = 6;
29    pub const SEND_RECEIPT: i32 = 7;
30    pub const SEND_ERROR: i32 = 8;
31    pub const MESSAGE: i32 = 9;
32    pub const ACK: i32 = 10;
33    pub const FLOW: i32 = 11;
34    pub const UNSUBSCRIBE: i32 = 12;
35    pub const SUCCESS: i32 = 13;
36    pub const ERROR: i32 = 14;
37    pub const CLOSE_PRODUCER: i32 = 15;
38    pub const CLOSE_CONSUMER: i32 = 16;
39    pub const PRODUCER_SUCCESS: i32 = 17;
40    pub const PING: i32 = 18;
41    pub const PONG: i32 = 19;
42    pub const REDELIVER_UNACKNOWLEDGED_MESSAGES: i32 = 20;
43    pub const PARTITIONED_METADATA: i32 = 21;
44    pub const PARTITIONED_METADATA_RESPONSE: i32 = 22;
45    pub const LOOKUP: i32 = 23;
46    pub const LOOKUP_RESPONSE: i32 = 24;
47    pub const GET_LAST_MESSAGE_ID: i32 = 29;
48    pub const GET_LAST_MESSAGE_ID_RESPONSE: i32 = 30;
49    pub const GET_SCHEMA: i32 = 34;
50    pub const GET_SCHEMA_RESPONSE: i32 = 35;
51}
52
53/// `ServerError` values NetGet answers with.
54pub mod server_error {
55    pub const UNKNOWN_ERROR: i32 = 0;
56    pub const AUTHORIZATION_ERROR: i32 = 4;
57    pub const SERVICE_NOT_READY: i32 = 6;
58    pub const CHECKSUM_ERROR: i32 = 9;
59    pub const UNSUPPORTED_VERSION_ERROR: i32 = 10;
60    pub const TOPIC_NOT_FOUND: i32 = 11;
61    pub const SUBSCRIPTION_NOT_FOUND: i32 = 12;
62    pub const CONSUMER_NOT_FOUND: i32 = 13;
63    pub const TOO_MANY_REQUESTS: i32 = 14;
64    pub const CONSUMER_BUSY: i32 = 5;
65    pub const INVALID_TOPIC_NAME: i32 = 17;
66    pub const NOT_ALLOWED_ERROR: i32 = 22;
67}
68
69pub const SUB_TYPES: [&str; 4] = ["Exclusive", "Shared", "Failover", "Key_Shared"];
70
71#[derive(Clone, PartialEq, Message)]
72pub struct KeyValue {
73    #[prost(string, required, tag = "1")]
74    pub key: String,
75    #[prost(string, required, tag = "2")]
76    pub value: String,
77}
78
79#[derive(Clone, PartialEq, Message)]
80pub struct MessageIdData {
81    #[prost(uint64, required, tag = "1")]
82    pub ledger_id: u64,
83    #[prost(uint64, required, tag = "2")]
84    pub entry_id: u64,
85    #[prost(int32, optional, tag = "3")]
86    pub partition: Option<i32>,
87    #[prost(int32, optional, tag = "4")]
88    pub batch_index: Option<i32>,
89}
90
91#[derive(Clone, PartialEq, Message)]
92pub struct MessageMetadata {
93    #[prost(string, required, tag = "1")]
94    pub producer_name: String,
95    #[prost(uint64, required, tag = "2")]
96    pub sequence_id: u64,
97    #[prost(uint64, required, tag = "3")]
98    pub publish_time: u64,
99    #[prost(message, repeated, tag = "4")]
100    pub properties: Vec<KeyValue>,
101    #[prost(string, optional, tag = "6")]
102    pub partition_key: Option<String>,
103    #[prost(int32, optional, tag = "8")]
104    pub compression: Option<i32>,
105    #[prost(uint32, optional, tag = "9")]
106    pub uncompressed_size: Option<u32>,
107    #[prost(int32, optional, tag = "11")]
108    pub num_messages_in_batch: Option<i32>,
109    #[prost(uint64, optional, tag = "12")]
110    pub event_time: Option<u64>,
111    #[prost(string, optional, tag = "14")]
112    pub encryption_algo: Option<String>,
113    #[prost(bool, optional, tag = "25")]
114    pub null_value: Option<bool>,
115    #[prost(int32, optional, tag = "27")]
116    pub num_chunks_from_msg: Option<i32>,
117}
118
119#[derive(Clone, PartialEq, Message)]
120pub struct SingleMessageMetadata {
121    #[prost(message, repeated, tag = "1")]
122    pub properties: Vec<KeyValue>,
123    #[prost(string, optional, tag = "2")]
124    pub partition_key: Option<String>,
125    #[prost(int32, required, tag = "3")]
126    pub payload_size: i32,
127    #[prost(bool, optional, tag = "4")]
128    pub compacted_out: Option<bool>,
129    #[prost(uint64, optional, tag = "5")]
130    pub event_time: Option<u64>,
131    #[prost(uint64, optional, tag = "8")]
132    pub sequence_id: Option<u64>,
133    #[prost(bool, optional, tag = "9")]
134    pub null_value: Option<bool>,
135}
136
137#[derive(Clone, PartialEq, Message)]
138pub struct CommandConnect {
139    #[prost(string, required, tag = "1")]
140    pub client_version: String,
141    #[prost(bytes = "vec", optional, tag = "3")]
142    pub auth_data: Option<Vec<u8>>,
143    #[prost(int32, optional, tag = "4")]
144    pub protocol_version: Option<i32>,
145    #[prost(string, optional, tag = "5")]
146    pub auth_method_name: Option<String>,
147    #[prost(string, optional, tag = "6")]
148    pub proxy_to_broker_url: Option<String>,
149}
150
151#[derive(Clone, PartialEq, Message)]
152pub struct CommandConnected {
153    #[prost(string, required, tag = "1")]
154    pub server_version: String,
155    #[prost(int32, optional, tag = "2")]
156    pub protocol_version: Option<i32>,
157    #[prost(int32, optional, tag = "3")]
158    pub max_message_size: Option<i32>,
159}
160
161#[derive(Clone, PartialEq, Message)]
162pub struct CommandSubscribe {
163    #[prost(string, required, tag = "1")]
164    pub topic: String,
165    #[prost(string, required, tag = "2")]
166    pub subscription: String,
167    #[prost(int32, required, tag = "3")]
168    pub sub_type: i32,
169    #[prost(uint64, required, tag = "4")]
170    pub consumer_id: u64,
171    #[prost(uint64, required, tag = "5")]
172    pub request_id: u64,
173    #[prost(string, optional, tag = "6")]
174    pub consumer_name: Option<String>,
175    #[prost(bool, optional, tag = "8")]
176    pub durable: Option<bool>,
177    #[prost(int32, optional, tag = "13")]
178    pub initial_position: Option<i32>,
179}
180
181#[derive(Clone, PartialEq, Message)]
182pub struct CommandPartitionedTopicMetadata {
183    #[prost(string, required, tag = "1")]
184    pub topic: String,
185    #[prost(uint64, required, tag = "2")]
186    pub request_id: u64,
187}
188
189#[derive(Clone, PartialEq, Message)]
190pub struct CommandPartitionedTopicMetadataResponse {
191    #[prost(uint32, optional, tag = "1")]
192    pub partitions: Option<u32>,
193    #[prost(uint64, required, tag = "2")]
194    pub request_id: u64,
195    #[prost(int32, optional, tag = "3")]
196    pub response: Option<i32>,
197    #[prost(int32, optional, tag = "4")]
198    pub error: Option<i32>,
199    #[prost(string, optional, tag = "5")]
200    pub message: Option<String>,
201}
202
203#[derive(Clone, PartialEq, Message)]
204pub struct CommandLookupTopic {
205    #[prost(string, required, tag = "1")]
206    pub topic: String,
207    #[prost(uint64, required, tag = "2")]
208    pub request_id: u64,
209    #[prost(bool, optional, tag = "3")]
210    pub authoritative: Option<bool>,
211}
212
213#[derive(Clone, PartialEq, Message)]
214pub struct CommandLookupTopicResponse {
215    #[prost(string, optional, tag = "1")]
216    pub broker_service_url: Option<String>,
217    #[prost(int32, optional, tag = "3")]
218    pub response: Option<i32>,
219    #[prost(uint64, required, tag = "4")]
220    pub request_id: u64,
221    #[prost(bool, optional, tag = "5")]
222    pub authoritative: Option<bool>,
223    #[prost(int32, optional, tag = "6")]
224    pub error: Option<i32>,
225    #[prost(string, optional, tag = "7")]
226    pub message: Option<String>,
227    #[prost(bool, optional, tag = "8")]
228    pub proxy_through_service_url: Option<bool>,
229}
230
231#[derive(Clone, PartialEq, Message)]
232pub struct CommandProducer {
233    #[prost(string, required, tag = "1")]
234    pub topic: String,
235    #[prost(uint64, required, tag = "2")]
236    pub producer_id: u64,
237    #[prost(uint64, required, tag = "3")]
238    pub request_id: u64,
239    #[prost(string, optional, tag = "4")]
240    pub producer_name: Option<String>,
241    #[prost(bool, optional, tag = "5")]
242    pub encrypted: Option<bool>,
243    #[prost(int32, optional, tag = "10")]
244    pub producer_access_mode: Option<i32>,
245}
246
247#[derive(Clone, PartialEq, Message)]
248pub struct CommandSend {
249    #[prost(uint64, required, tag = "1")]
250    pub producer_id: u64,
251    #[prost(uint64, required, tag = "2")]
252    pub sequence_id: u64,
253    #[prost(int32, optional, tag = "3")]
254    pub num_messages: Option<i32>,
255    #[prost(uint64, optional, tag = "4")]
256    pub txnid_least_bits: Option<u64>,
257    #[prost(uint64, optional, tag = "6")]
258    pub highest_sequence_id: Option<u64>,
259    #[prost(bool, optional, tag = "7")]
260    pub is_chunk: Option<bool>,
261}
262
263#[derive(Clone, PartialEq, Message)]
264pub struct CommandSendReceipt {
265    #[prost(uint64, required, tag = "1")]
266    pub producer_id: u64,
267    #[prost(uint64, required, tag = "2")]
268    pub sequence_id: u64,
269    #[prost(message, optional, tag = "3")]
270    pub message_id: Option<MessageIdData>,
271    #[prost(uint64, optional, tag = "4")]
272    pub highest_sequence_id: Option<u64>,
273}
274
275#[derive(Clone, PartialEq, Message)]
276pub struct CommandSendError {
277    #[prost(uint64, required, tag = "1")]
278    pub producer_id: u64,
279    #[prost(uint64, required, tag = "2")]
280    pub sequence_id: u64,
281    #[prost(int32, required, tag = "3")]
282    pub error: i32,
283    #[prost(string, required, tag = "4")]
284    pub message: String,
285}
286
287#[derive(Clone, PartialEq, Message)]
288pub struct CommandMessage {
289    #[prost(uint64, required, tag = "1")]
290    pub consumer_id: u64,
291    #[prost(message, required, tag = "2")]
292    pub message_id: MessageIdData,
293    #[prost(uint32, optional, tag = "3")]
294    pub redelivery_count: Option<u32>,
295}
296
297#[derive(Clone, PartialEq, Message)]
298pub struct CommandAck {
299    #[prost(uint64, required, tag = "1")]
300    pub consumer_id: u64,
301    #[prost(int32, required, tag = "2")]
302    pub ack_type: i32,
303    #[prost(message, repeated, tag = "3")]
304    pub message_id: Vec<MessageIdData>,
305    #[prost(uint64, optional, tag = "8")]
306    pub request_id: Option<u64>,
307}
308
309#[derive(Clone, PartialEq, Message)]
310pub struct CommandAckResponse {
311    #[prost(uint64, required, tag = "1")]
312    pub consumer_id: u64,
313    #[prost(uint64, optional, tag = "6")]
314    pub request_id: Option<u64>,
315}
316
317#[derive(Clone, PartialEq, Message)]
318pub struct CommandFlow {
319    #[prost(uint64, required, tag = "1")]
320    pub consumer_id: u64,
321    #[prost(uint32, required, tag = "2")]
322    pub message_permits: u32,
323}
324
325#[derive(Clone, PartialEq, Message)]
326pub struct CommandUnsubscribe {
327    #[prost(uint64, required, tag = "1")]
328    pub consumer_id: u64,
329    #[prost(uint64, required, tag = "2")]
330    pub request_id: u64,
331}
332
333#[derive(Clone, PartialEq, Message)]
334pub struct CommandCloseProducer {
335    #[prost(uint64, required, tag = "1")]
336    pub producer_id: u64,
337    #[prost(uint64, required, tag = "2")]
338    pub request_id: u64,
339}
340
341#[derive(Clone, PartialEq, Message)]
342pub struct CommandCloseConsumer {
343    #[prost(uint64, required, tag = "1")]
344    pub consumer_id: u64,
345    #[prost(uint64, required, tag = "2")]
346    pub request_id: u64,
347}
348
349#[derive(Clone, PartialEq, Message)]
350pub struct CommandRedeliverUnacknowledgedMessages {
351    #[prost(uint64, required, tag = "1")]
352    pub consumer_id: u64,
353    #[prost(message, repeated, tag = "2")]
354    pub message_ids: Vec<MessageIdData>,
355}
356
357#[derive(Clone, PartialEq, Message)]
358pub struct CommandSuccess {
359    #[prost(uint64, required, tag = "1")]
360    pub request_id: u64,
361}
362
363#[derive(Clone, PartialEq, Message)]
364pub struct CommandProducerSuccess {
365    #[prost(uint64, required, tag = "1")]
366    pub request_id: u64,
367    #[prost(string, required, tag = "2")]
368    pub producer_name: String,
369    #[prost(int64, optional, tag = "3")]
370    pub last_sequence_id: Option<i64>,
371    /// The Java client reads this unconditionally; a broker without schemas sends it empty.
372    #[prost(bytes = "vec", optional, tag = "4")]
373    pub schema_version: Option<Vec<u8>>,
374    #[prost(bool, optional, tag = "6")]
375    pub producer_ready: Option<bool>,
376}
377
378#[derive(Clone, PartialEq, Message)]
379pub struct CommandError {
380    #[prost(uint64, required, tag = "1")]
381    pub request_id: u64,
382    #[prost(int32, required, tag = "2")]
383    pub error: i32,
384    #[prost(string, required, tag = "3")]
385    pub message: String,
386}
387
388#[derive(Clone, PartialEq, Message)]
389pub struct CommandPing {}
390
391#[derive(Clone, PartialEq, Message)]
392pub struct CommandPong {}
393
394#[derive(Clone, PartialEq, Message)]
395pub struct CommandGetLastMessageId {
396    #[prost(uint64, required, tag = "1")]
397    pub consumer_id: u64,
398    #[prost(uint64, required, tag = "2")]
399    pub request_id: u64,
400}
401
402#[derive(Clone, PartialEq, Message)]
403pub struct CommandGetLastMessageIdResponse {
404    #[prost(message, required, tag = "1")]
405    pub last_message_id: MessageIdData,
406    #[prost(uint64, required, tag = "2")]
407    pub request_id: u64,
408}
409
410#[derive(Clone, PartialEq, Message)]
411pub struct CommandGetSchema {
412    #[prost(uint64, required, tag = "1")]
413    pub request_id: u64,
414    #[prost(string, required, tag = "2")]
415    pub topic: String,
416}
417
418#[derive(Clone, PartialEq, Message)]
419pub struct CommandGetSchemaResponse {
420    #[prost(uint64, required, tag = "1")]
421    pub request_id: u64,
422    #[prost(int32, optional, tag = "2")]
423    pub error_code: Option<i32>,
424    #[prost(string, optional, tag = "3")]
425    pub error_message: Option<String>,
426}
427
428#[derive(Clone, PartialEq, Message)]
429pub struct BaseCommand {
430    #[prost(int32, required, tag = "1")]
431    pub r#type: i32,
432    #[prost(message, optional, tag = "2")]
433    pub connect: Option<CommandConnect>,
434    #[prost(message, optional, tag = "3")]
435    pub connected: Option<CommandConnected>,
436    #[prost(message, optional, tag = "4")]
437    pub subscribe: Option<CommandSubscribe>,
438    #[prost(message, optional, tag = "5")]
439    pub producer: Option<CommandProducer>,
440    #[prost(message, optional, tag = "6")]
441    pub send: Option<CommandSend>,
442    #[prost(message, optional, tag = "7")]
443    pub send_receipt: Option<CommandSendReceipt>,
444    #[prost(message, optional, tag = "8")]
445    pub send_error: Option<CommandSendError>,
446    #[prost(message, optional, tag = "9")]
447    pub message: Option<CommandMessage>,
448    #[prost(message, optional, tag = "10")]
449    pub ack: Option<CommandAck>,
450    #[prost(message, optional, tag = "11")]
451    pub flow: Option<CommandFlow>,
452    #[prost(message, optional, tag = "12")]
453    pub unsubscribe: Option<CommandUnsubscribe>,
454    #[prost(message, optional, tag = "13")]
455    pub success: Option<CommandSuccess>,
456    #[prost(message, optional, tag = "14")]
457    pub error: Option<CommandError>,
458    #[prost(message, optional, tag = "15")]
459    pub close_producer: Option<CommandCloseProducer>,
460    #[prost(message, optional, tag = "16")]
461    pub close_consumer: Option<CommandCloseConsumer>,
462    #[prost(message, optional, tag = "17")]
463    pub producer_success: Option<CommandProducerSuccess>,
464    #[prost(message, optional, tag = "18")]
465    pub ping: Option<CommandPing>,
466    #[prost(message, optional, tag = "19")]
467    pub pong: Option<CommandPong>,
468    #[prost(message, optional, tag = "20")]
469    pub redeliver_unacknowledged_messages: Option<CommandRedeliverUnacknowledgedMessages>,
470    #[prost(message, optional, tag = "21")]
471    pub partition_metadata: Option<CommandPartitionedTopicMetadata>,
472    #[prost(message, optional, tag = "22")]
473    pub partition_metadata_response: Option<CommandPartitionedTopicMetadataResponse>,
474    #[prost(message, optional, tag = "23")]
475    pub lookup_topic: Option<CommandLookupTopic>,
476    #[prost(message, optional, tag = "24")]
477    pub lookup_topic_response: Option<CommandLookupTopicResponse>,
478    #[prost(message, optional, tag = "29")]
479    pub get_last_message_id: Option<CommandGetLastMessageId>,
480    #[prost(message, optional, tag = "30")]
481    pub get_last_message_id_response: Option<CommandGetLastMessageIdResponse>,
482    #[prost(message, optional, tag = "34")]
483    pub get_schema: Option<CommandGetSchema>,
484    #[prost(message, optional, tag = "35")]
485    pub get_schema_response: Option<CommandGetSchemaResponse>,
486    #[prost(message, optional, tag = "38")]
487    pub ack_response: Option<CommandAckResponse>,
488}
489
490impl BaseCommand {
491    pub fn of(kind: i32) -> Self {
492        BaseCommand {
493            r#type: kind,
494            ..Default::default()
495        }
496    }
497}
498
499/// CRC-32C (Castagnoli), as Pulsar checksums metadata and payload.
500pub fn crc32c(data: &[u8]) -> u32 {
501    const TABLE: [u32; 256] = {
502        let mut t = [0u32; 256];
503        let mut i = 0;
504        while i < 256 {
505            let mut c = i as u32;
506            let mut k = 0;
507            while k < 8 {
508                c = if c & 1 != 0 {
509                    0x82F6_3B78 ^ (c >> 1)
510                } else {
511                    c >> 1
512                };
513                k += 1;
514            }
515            t[i] = c;
516            i += 1;
517        }
518        t
519    };
520    let mut crc = !0u32;
521    for b in data {
522        crc = TABLE[((crc ^ u32::from(*b)) & 0xff) as usize] ^ (crc >> 8);
523    }
524    !crc
525}
526
527/// A frame carrying only a command.
528pub fn simple(cmd: &BaseCommand) -> Vec<u8> {
529    let body = cmd.encode_to_vec();
530    let mut out = Vec::with_capacity(8 + body.len());
531    out.extend_from_slice(&((4 + body.len()) as u32).to_be_bytes());
532    out.extend_from_slice(&(body.len() as u32).to_be_bytes());
533    out.extend_from_slice(&body);
534    out
535}
536
537/// A frame carrying a command, then the checksummed metadata and payload (SEND, MESSAGE).
538pub fn with_payload(cmd: &BaseCommand, metadata: &MessageMetadata, payload: &[u8]) -> Vec<u8> {
539    let body = cmd.encode_to_vec();
540    let meta = metadata.encode_to_vec();
541    let mut checked = Vec::with_capacity(4 + meta.len() + payload.len());
542    checked.extend_from_slice(&(meta.len() as u32).to_be_bytes());
543    checked.extend_from_slice(&meta);
544    checked.extend_from_slice(payload);
545    let total = 4 + body.len() + 2 + 4 + checked.len();
546    let mut out = Vec::with_capacity(4 + total);
547    out.extend_from_slice(&(total as u32).to_be_bytes());
548    out.extend_from_slice(&(body.len() as u32).to_be_bytes());
549    out.extend_from_slice(&body);
550    out.extend_from_slice(&MAGIC_CRC32C.to_be_bytes());
551    out.extend_from_slice(&crc32c(&checked).to_be_bytes());
552    out.extend_from_slice(&checked);
553    out
554}
555
556/// One frame read off the wire.
557pub struct Frame {
558    pub command: BaseCommand,
559    /// For SEND and MESSAGE: the metadata and the (possibly batched) payload.
560    pub payload: Option<(MessageMetadata, Vec<u8>)>,
561}
562
563/// A frame's payload section: none, a decoded one, or why it was refused.
564pub type Payload = Result<Option<(MessageMetadata, Vec<u8>)>, PayloadError>;
565
566/// Why a payload section was refused; the frame itself was well-formed.
567#[derive(Debug)]
568pub enum PayloadError {
569    Checksum,
570    Malformed(String),
571}
572
573/// Read one frame. A frame over `MAX_FRAME` or a command that does not decode is an error:
574/// the peer is not speaking Pulsar. A bad checksum or metadata is returned beside the command,
575/// so a SEND can be answered with a SendError rather than dropping the connection.
576pub async fn read_frame<R: AsyncRead + Unpin>(
577    r: &mut R,
578    idle: Duration,
579) -> Result<Option<(BaseCommand, Payload)>> {
580    let mut size = [0u8; 4];
581    match tokio::time::timeout(idle, r.read_exact(&mut size)).await {
582        Err(_) => bail!("Pulsar peer idle"),
583        Ok(Err(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
584        Ok(r) => {
585            r?;
586        }
587    }
588    let total = u32::from_be_bytes(size) as usize;
589    ensure!(
590        (4..=MAX_FRAME).contains(&total),
591        "Pulsar frame of {total} bytes (at most {MAX_FRAME})"
592    );
593    let mut frame = vec![0u8; total];
594    tokio::time::timeout(IO_TIMEOUT, r.read_exact(&mut frame))
595        .await
596        .context("Pulsar frame read deadline")??;
597    let cmd_len = u32::from_be_bytes(frame[0..4].try_into().unwrap_or_default()) as usize;
598    ensure!(
599        cmd_len <= MAX_COMMAND && 4 + cmd_len <= total,
600        "Pulsar command of {cmd_len} bytes does not fit"
601    );
602    let command = BaseCommand::decode(&frame[4..4 + cmd_len])
603        .context("Pulsar command is not a BaseCommand")?;
604    let rest = &frame[4 + cmd_len..];
605    if rest.is_empty() {
606        return Ok(Some((command, Ok(None))));
607    }
608    Ok(Some((command, parse_payload(rest))))
609}
610
611fn parse_payload(rest: &[u8]) -> Payload {
612    let bad = |m: &str| PayloadError::Malformed(m.to_string());
613    let checked = if rest.len() >= 6 && u16::from_be_bytes([rest[0], rest[1]]) == MAGIC_CRC32C {
614        let want = u32::from_be_bytes([rest[2], rest[3], rest[4], rest[5]]);
615        let checked = &rest[6..];
616        if crc32c(checked) != want {
617            return Err(PayloadError::Checksum);
618        }
619        checked
620    } else {
621        rest
622    };
623    if checked.len() < 4 {
624        return Err(bad("payload section too short"));
625    }
626    let meta_len = u32::from_be_bytes([checked[0], checked[1], checked[2], checked[3]]) as usize;
627    if 4 + meta_len > checked.len() {
628        return Err(bad("metadata size past the frame"));
629    }
630    let metadata = MessageMetadata::decode(&checked[4..4 + meta_len])
631        .map_err(|e| bad(&format!("metadata: {e}")))?;
632    Ok(Some((metadata, checked[4 + meta_len..].to_vec())))
633}
634
635/// One message out of a payload: what a single message or one batch entry carries.
636#[derive(Clone, Debug)]
637pub struct Entry {
638    pub properties: Vec<KeyValue>,
639    pub key: Option<String>,
640    pub event_time: Option<u64>,
641    pub sequence_id: u64,
642    pub payload: Vec<u8>,
643}
644
645/// The messages a SEND carries: one, or each entry of a batch. Compressed and encrypted
646/// payloads are refused (NetGet decompresses nothing it could not show the model).
647pub fn entries(meta: &MessageMetadata, payload: &[u8]) -> Result<Vec<Entry>> {
648    ensure!(
649        meta.compression.unwrap_or(0) == 0,
650        "compressed payloads are not supported"
651    );
652    ensure!(
653        meta.encryption_algo.is_none(),
654        "encrypted payloads are not supported"
655    );
656    ensure!(
657        meta.num_chunks_from_msg.unwrap_or(1) <= 1,
658        "chunked messages are not supported"
659    );
660    let Some(n) = meta.num_messages_in_batch.filter(|_| batched(meta)) else {
661        return Ok(vec![Entry {
662            properties: meta.properties.clone(),
663            key: meta.partition_key.clone(),
664            event_time: meta.event_time.filter(|t| *t > 0),
665            sequence_id: meta.sequence_id,
666            payload: payload.to_vec(),
667        }]);
668    };
669    ensure!(
670        (1..=MAX_BATCH as i32).contains(&n),
671        "a batch of {n} messages"
672    );
673    let mut out = Vec::new();
674    let mut rest = payload;
675    for i in 0..n as u64 {
676        ensure!(rest.len() >= 4, "batch entry {i} is cut short");
677        let len = u32::from_be_bytes([rest[0], rest[1], rest[2], rest[3]]) as usize;
678        ensure!(
679            4 + len <= rest.len(),
680            "batch entry {i} metadata past the payload"
681        );
682        let single =
683            SingleMessageMetadata::decode(&rest[4..4 + len]).context("batch entry metadata")?;
684        rest = &rest[4 + len..];
685        let size = usize::try_from(single.payload_size).context("negative payload size")?;
686        ensure!(
687            size <= rest.len(),
688            "batch entry {i} payload past the payload"
689        );
690        let (body, tail) = rest.split_at(size);
691        rest = tail;
692        if single.compacted_out == Some(true) {
693            continue;
694        }
695        out.push(Entry {
696            properties: single.properties,
697            key: single.partition_key,
698            event_time: single.event_time.filter(|t| *t > 0),
699            sequence_id: single.sequence_id.unwrap_or(meta.sequence_id + i),
700            payload: body.to_vec(),
701        });
702    }
703    Ok(out)
704}
705
706/// Whether the producer batched: clients that batch always set num_messages_in_batch.
707fn batched(meta: &MessageMetadata) -> bool {
708    meta.num_messages_in_batch.is_some()
709}
710
711/// A topic name in its full form: `persistent://public/default/t` from `t`.
712pub fn full_topic(t: &str) -> Result<String> {
713    ensure!(
714        !t.is_empty() && t.len() <= 512,
715        "topic name must be 1-512 bytes"
716    );
717    if t.starts_with("persistent://") || t.starts_with("non-persistent://") {
718        let rest = t.split_once("://").map(|(_, r)| r).unwrap_or_default();
719        ensure!(
720            rest.split('/').count() == 3 && rest.split('/').all(|p| !p.is_empty()),
721            "topic must be tenant/namespace/name"
722        );
723        return Ok(t.to_string());
724    }
725    ensure!(!t.contains("://"), "unknown topic domain");
726    Ok(match t.split('/').count() {
727        1 => format!("persistent://public/default/{t}"),
728        3 => format!("persistent://{t}"),
729        _ => bail!("topic must be a name or tenant/namespace/name"),
730    })
731}
732
733/// Text for the model: UTF-8 as is, anything else as hex, and which it is.
734pub fn show(payload: &[u8]) -> (String, &'static str) {
735    match std::str::from_utf8(payload) {
736        Ok(s)
737            if !s
738                .chars()
739                .any(|c| c.is_control() && !matches!(c, '\n' | '\r' | '\t')) =>
740        {
741            (s.to_string(), "utf8")
742        }
743        _ => (hex::encode(payload), "hex"),
744    }
745}
746
747/// Bytes from the model: `encoding` utf8 (default) or hex.
748pub fn bytes(payload: &str, encoding: Option<&str>) -> Result<Vec<u8>> {
749    match encoding.unwrap_or("utf8") {
750        "utf8" => Ok(payload.as_bytes().to_vec()),
751        "hex" => hex::decode(payload).context("payload is not hex"),
752        other => bail!("encoding must be utf8 or hex, not {other}"),
753    }
754}
755
756pub fn now_millis() -> u64 {
757    crate::utils::clock::SystemTime::now()
758        .duration_since(crate::utils::clock::UNIX_EPOCH)
759        .map(|d| d.as_millis() as u64)
760        .unwrap_or(0)
761}
netget/server/pulsar/wire.rs

netget/server/pulsar/
wire.rs
