netget/server/nsq/
actions.rs

1//! NSQ actions: what the model is told, and how its answers become frames.
2//!
3//! The model is the broker's judgement: it accepts or refuses publishes and subscriptions, and
4//! it decides which messages a subscriber receives. It never writes a frame. Every action here
5//! returns a structured answer (`ActionResult::Custom "nsq_answer"`) that the session loop
6//! renders through [`super::wire`] — the loop, not the model, assigns message ids and
7//! timestamps and holds deliveries to the client's RDY count.
8
9use super::wire;
10use crate::llm::actions::{
11    protocol_trait::{ActionResult, Protocol, Server},
12    ActionDefinition, Parameter,
13};
14use crate::protocol::log_template::LogTemplate;
15use crate::protocol::EventType;
16use crate::state::app_state::AppState;
17use anyhow::{anyhow, Context, Result};
18use serde_json::{json, Value};
19use std::sync::LazyLock;
20
21pub struct NsqProtocol;
22
23impl NsqProtocol {
24    pub fn new() -> Self {
25        Self
26    }
27}
28
29impl Default for NsqProtocol {
30    fn default() -> Self {
31        Self::new()
32    }
33}
34
35/// The name of every structured answer this protocol's executor returns.
36pub const ANSWER: &str = "nsq_answer";
37
38/// One answer, read back from what [`NsqProtocol::execute_action`] returned.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub enum Answer {
41    Ok,
42    Error { code: String, message: String },
43    Deliver(Vec<Delivery>),
44    Close,
45}
46
47/// A message the model wants delivered: its body and the attempts count the frame carries.
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct Delivery {
50    pub body: String,
51    pub attempts: u16,
52}
53
54impl Answer {
55    /// Read an executor result back. `None` for anything that is not an NSQ answer.
56    pub fn from_result(result: &ActionResult) -> Option<Answer> {
57        match result {
58            ActionResult::CloseConnection => Some(Answer::Close),
59            ActionResult::Custom { name, data } if name == ANSWER => {
60                match data.get("kind")?.as_str()? {
61                    "ok" => Some(Answer::Ok),
62                    "error" => Some(Answer::Error {
63                        code: data.get("code")?.as_str()?.to_string(),
64                        message: data
65                            .get("message")
66                            .and_then(Value::as_str)
67                            .unwrap_or("")
68                            .to_string(),
69                    }),
70                    "deliver" => Some(Answer::Deliver(
71                        data.get("messages")?
72                            .as_array()?
73                            .iter()
74                            .filter_map(|m| {
75                                Some(Delivery {
76                                    body: m.get("body")?.as_str()?.to_string(),
77                                    attempts: m
78                                        .get("attempts")
79                                        .and_then(Value::as_u64)
80                                        .and_then(|a| u16::try_from(a).ok())
81                                        .unwrap_or(1),
82                                })
83                            })
84                            .collect(),
85                    )),
86                    _ => None,
87                }
88            }
89            _ => None,
90        }
91    }
92}
93
94impl Protocol for NsqProtocol {
95    fn get_startup_parameters(&self) -> Vec<crate::llm::actions::ParameterDefinition> {
96        vec![
97            crate::llm::actions::ParameterDefinition {
98                name: "first_byte_timeout_secs".to_string(),
99                type_hint: "number".to_string(),
100                description: "Seconds a new connection may send nothing (not even the '  V2' \
101                              magic) before the server closes it. Default 300, the window a \
102                              `manual` rule gives a human - the peer may be NetGet's own TCP \
103                              client parked for its operator."
104                    .to_string(),
105                required: false,
106                example: json!(300),
107                default: Some(json!(super::FIRST_BYTE_TIMEOUT.as_secs())),
108            },
109            crate::llm::actions::ParameterDefinition {
110                name: "heartbeat_interval_secs".to_string(),
111                type_hint: "number".to_string(),
112                description: "Seconds between the _heartbeat_ frames sent to a client that did \
113                              not ask for its own interval in IDENTIFY. A client silent for two \
114                              intervals is closed, as nsqd does. Default 30 (nsqd's)."
115                    .to_string(),
116                required: false,
117                example: json!(30),
118                default: Some(json!(super::HEARTBEAT_INTERVAL.as_secs())),
119            },
120            crate::llm::actions::ParameterDefinition {
121                name: "idle_timeout_secs".to_string(),
122                type_hint: "number".to_string(),
123                description: "Seconds of silence after which a client that disabled heartbeats \
124                              (heartbeat_interval -1 in IDENTIFY) is closed. Default 300. A \
125                              client waiting for the model's answer is not idle."
126                    .to_string(),
127                required: false,
128                example: json!(300),
129                default: Some(json!(super::IDLE_TIMEOUT.as_secs())),
130            },
131        ]
132    }
133    fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
134        Vec::new()
135    }
136    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
137        vec![
138            ok_action(),
139            error_action(),
140            deliver_action(),
141            close_connection_action(),
142        ]
143    }
144    fn protocol_name(&self) -> &'static str {
145        "NSQ"
146    }
147    fn get_event_types(&self) -> Vec<EventType> {
148        vec![
149            NSQ_PUBLISH_EVENT.clone(),
150            NSQ_SUBSCRIBE_EVENT.clone(),
151            NSQ_READY_EVENT.clone(),
152            NSQ_FINISH_EVENT.clone(),
153            NSQ_REQUEUE_EVENT.clone(),
154        ]
155    }
156    fn stack_name(&self) -> &'static str {
157        "ETH>IP>TCP>NSQ"
158    }
159    fn keywords(&self) -> Vec<&'static str> {
160        vec!["nsq", "nsqd", "message queue", "nsq_tail", "to_nsq"]
161    }
162    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
163        use crate::protocol::metadata::{
164            DevelopmentState, PrivilegeRequirement, ProtocolMetadataV2,
165        };
166
167        ProtocolMetadataV2::builder()
168            // Beta on evidence: tests/server/nsq/real_client_test.rs drives the NSQ project's
169            // own to_nsq and nsq_tail (go-nsq) and hard-fails when they are absent. Not Stable:
170            // both are one implementation (go-nsq), there is no pcap oracle (Wireshark has no
171            // NSQ dissector), and in-flight timeouts are not implemented.
172            .state(DevelopmentState::Beta)
173            .well_known_port(4150)
174            // 4150 is unprivileged, and so is every port a test picks.
175            .privilege_requirement(PrivilegeRequirement::None)
176            .implementation(
177                "Hand-written nsqd TCP protocol (V2) over tokio: the '  V2' magic, \
178                 newline-terminated commands with size-prefixed bodies, and size/type framed \
179                 responses, errors and messages, all rendered by NetGet; message ids and \
180                 timestamps generated by NetGet",
181            )
182            .llm_control(
183                "Whether each PUB/MPUB/DPUB and SUB is accepted (OK) or refused (an nsqd error \
184                 code), and which messages a subscriber receives when it signals RDY, finishes \
185                 or requeues one",
186            )
187            .e2e_testing(
188                "tests/server/nsq/real_client_test.rs drives the NSQ project's own to_nsq \
189                 (publishes stdin lines) and nsq_tail (subscribes and prints what it receives) - \
190                 go-nsq, Go; Homebrew `nsq`, the release tarball on Ubuntu - asserting that \
191                 to_nsq's lines reach the handler as nsq_publish and that nsq_tail prints exactly \
192                 the bodies delivered. It fails, never skips, when the binaries are absent. \
193                 tests/server/nsq/e2e_test.rs covers the mocked-model path on a raw socket.",
194            )
195            .notes(
196                "Implements the V2 magic; IDENTIFY (plain OK, or the feature-negotiation JSON \
197                 with TLS, deflate, snappy and auth off); SUB; RDY; FIN; REQ; TOUCH; PUB, MPUB \
198                 and DPUB; NOP; CLS (CLOSE_WAIT); heartbeats (_heartbeat_ on the negotiated \
199                 interval, and a client silent for two intervals is closed). AUTH answers \
200                 E_AUTH_DISABLED. NetGet is not a queue: nothing published is stored or routed \
201                 to other connections - the model decides what each subscriber receives. \
202                 In-flight messages never time out and are not redelivered unless the model \
203                 redelivers a requeued one. Bodies are capped at nsqd's defaults (1 MiB per \
204                 message, 5 MiB per MPUB/IDENTIFY body), judged from the declared size before \
205                 allocation. On backend failure a PUB gets E_PUB_FAILED and a SUB E_INVALID, \
206                 both with a fixed text.",
207            )
208            .max_inbound_bytes(wire::MAX_BODY_SIZE)
209            // PUB/MPUB/DPUB get E_*PUB_FAILED and SUB gets E_INVALID; RDY, FIN and REQ take no
210            // reply in nsqd, so a failure there delivers nothing, which is the honest answer.
211            .answers_on_failure()
212            .build()
213    }
214    fn description(&self) -> &'static str {
215        "NSQ message broker (nsqd TCP protocol) - the model decides what is accepted and delivered"
216    }
217    fn example_prompt(&self) -> &'static str {
218        "NSQ broker on port 4150 - accept every publish, and give subscribers of topic 'orders' \
219         the message 'order 1 shipped'"
220    }
221    fn group_name(&self) -> &'static str {
222        "Application"
223    }
224    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
225        use crate::llm::actions::StartupExamples;
226        StartupExamples::new(
227            json!({
228                "type": "open_server",
229                "port": 4150,
230                "base_stack": "nsq",
231                "instruction": "NSQ broker. Accept every publish and every subscription. The \
232                                topic 'orders' holds two messages: 'order 1 shipped' and \
233                                'order 2 packed'. Deliver them once each to a subscriber."
234            }),
235            json!({
236                "type": "open_server",
237                "port": 4150,
238                "base_stack": "nsq",
239                "event_handlers": [{
240                    "event_pattern": "*",
241                    "handler": {
242                        "type": "script",
243                        "language": "python",
244                        "code": "import json, sys\ni = json.load(sys.stdin)\nt = i['event_type_id']\ne = i['event']\nif t in ('nsq_publish', 'nsq_subscribe'):\n    a = [{'type': 'send_nsq_ok'}]\nelif t == 'nsq_ready':\n    a = [{'type': 'deliver_nsq_messages', 'messages': [{'body': 'hello from ' + e.get('topic', '')}]}]\nelse:\n    a = []\nprint(json.dumps({'actions': a}))"
245                    }
246                }]
247            }),
248            json!({
249                "type": "open_server",
250                "port": 4150,
251                "base_stack": "nsq",
252                "event_handlers": [
253                    {
254                        "event_pattern": "nsq_publish",
255                        "handler": {"type": "static", "actions": [{"type": "send_nsq_ok"}]}
256                    },
257                    {
258                        "event_pattern": "nsq_subscribe",
259                        "handler": {"type": "static", "actions": [{"type": "send_nsq_ok"}]}
260                    },
261                    {
262                        "event_pattern": "nsq_ready",
263                        "handler": {"type": "static", "actions": [
264                            {"type": "deliver_nsq_messages", "messages": [{"body": "hello"}]}
265                        ]}
266                    }
267                ]
268            }),
269        )
270    }
271}
272
273impl Server for NsqProtocol {
274    fn spawn(
275        &self,
276        ctx: crate::protocol::SpawnContext,
277    ) -> std::pin::Pin<
278        Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
279    > {
280        Box::pin(async move {
281            let secs = |name: &str| -> anyhow::Result<Option<u64>> {
282                Ok(ctx
283                    .startup_params
284                    .as_ref()
285                    .map(|p| p.get_optional_u64(name))
286                    .transpose()?
287                    .flatten())
288            };
289            let config = super::NsqConfig {
290                first_byte_timeout: secs("first_byte_timeout_secs")?
291                    .map(std::time::Duration::from_secs)
292                    .unwrap_or(super::FIRST_BYTE_TIMEOUT),
293                heartbeat_interval: secs("heartbeat_interval_secs")?
294                    .map(|s| std::time::Duration::from_secs(s.max(1)))
295                    .unwrap_or(super::HEARTBEAT_INTERVAL),
296                idle_timeout: secs("idle_timeout_secs")?
297                    .map(std::time::Duration::from_secs)
298                    .unwrap_or(super::IDLE_TIMEOUT),
299            };
300
301            crate::server::nsq::NsqServer::spawn_with_llm_actions(
302                ctx.legacy_listen_addr(),
303                ctx.llm_client,
304                ctx.state,
305                ctx.status_tx,
306                ctx.server_id,
307                config,
308            )
309            .await
310        })
311    }
312
313    fn execute_action(&self, action: Value) -> Result<ActionResult> {
314        let action_type = action
315            .get("type")
316            .and_then(|v| v.as_str())
317            .context("Missing 'type' field in action")?;
318
319        let data = match action_type {
320            "send_nsq_ok" => json!({"kind": "ok"}),
321            "send_nsq_error" => {
322                let code = action
323                    .get("code")
324                    .and_then(Value::as_str)
325                    .context("send_nsq_error needs 'code'")?
326                    .trim()
327                    .to_ascii_uppercase();
328                if !wire::ERROR_CODES.contains(&code.as_str()) {
329                    return Err(anyhow!(
330                        "send_nsq_error: unknown code {code:?}; use one of {}",
331                        wire::ERROR_CODES.join(", ")
332                    ));
333                }
334                let message = action.get("message").and_then(Value::as_str).unwrap_or("");
335                let message = crate::utils::truncate_for_log(
336                    &crate::utils::sanitize::line_field(message),
337                    256,
338                );
339                json!({"kind": "error", "code": code, "message": message})
340            }
341            "deliver_nsq_messages" => {
342                let messages = action
343                    .get("messages")
344                    .and_then(Value::as_array)
345                    .context("deliver_nsq_messages needs 'messages', a list of {body}")?;
346                let mut out = Vec::with_capacity(messages.len());
347                for (i, m) in messages.iter().enumerate() {
348                    let body = match m {
349                        Value::String(s) => s.clone(),
350                        Value::Object(o) => match o.get("body") {
351                            Some(Value::String(s)) => s.clone(),
352                            Some(Value::Null) | None => {
353                                return Err(anyhow!("message {i} has no 'body'"))
354                            }
355                            Some(other) => other.to_string(),
356                        },
357                        other => other.to_string(),
358                    };
359                    if body.len() > wire::MAX_MSG_SIZE {
360                        return Err(anyhow!(
361                            "message {i} is {} bytes, over the {} byte limit",
362                            body.len(),
363                            wire::MAX_MSG_SIZE
364                        ));
365                    }
366                    let attempts = match m.get("attempts") {
367                        None | Some(Value::Null) => 1,
368                        Some(v) => v
369                            .as_u64()
370                            .or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
371                            .filter(|a| (1..=u16::MAX as u64).contains(a))
372                            .with_context(|| {
373                                format!("message {i}: 'attempts' must be 1 to 65535")
374                            })?,
375                    };
376                    out.push(json!({"body": body, "attempts": attempts}));
377                }
378                json!({"kind": "deliver", "messages": out})
379            }
380            "close_connection" => return Ok(ActionResult::CloseConnection),
381            _ => return Err(anyhow!("Unknown NSQ action: {}", action_type)),
382        };
383        Ok(ActionResult::Custom {
384            name: ANSWER.to_string(),
385            data,
386        })
387    }
388}
389
390fn ok_action() -> ActionDefinition {
391    ActionDefinition {
392        name: "send_nsq_ok".to_string(),
393        description: "Accept a publish (PUB, MPUB, DPUB) or a subscription (SUB): the client \
394                      receives OK."
395            .to_string(),
396        parameters: vec![],
397        example: json!({"type": "send_nsq_ok"}),
398        log_template: Some(
399            LogTemplate::new()
400                .with_info("-> NSQ OK")
401                .with_debug("NSQ send_nsq_ok"),
402        ),
403    }
404}
405
406fn error_action() -> ActionDefinition {
407    ActionDefinition {
408        name: "send_nsq_error".to_string(),
409        description: "Refuse the request with one of nsqd's error codes. E_PUB_FAILED, \
410                      E_MPUB_FAILED or E_DPUB_FAILED refuse a publish; E_BAD_TOPIC, \
411                      E_BAD_CHANNEL, E_UNAUTHORIZED or E_TOO_MANY_CHANNEL_CONSUMERS refuse a \
412                      subscription. Every code except E_FIN_FAILED, E_REQ_FAILED and \
413                      E_TOUCH_FAILED closes the connection, as nsqd does."
414            .to_string(),
415        parameters: vec![
416            Parameter {
417                name: "code".to_string(),
418                type_hint: "string".to_string(),
419                description: "The nsqd error code".to_string(),
420                required: true,
421            }
422            .with_choices(wire::ERROR_CODES.iter().copied()),
423            Parameter {
424                name: "message".to_string(),
425                type_hint: "string".to_string(),
426                description: "One line saying why, after the code".to_string(),
427                required: false,
428            },
429        ],
430        example: json!({"type": "send_nsq_error", "code": "E_PUB_FAILED", "message": "<why>"}),
431        log_template: Some(
432            LogTemplate::new()
433                .with_info("-> NSQ {code}")
434                .with_debug("NSQ send_nsq_error: code={code}"),
435        ),
436    }
437}
438
439fn deliver_action() -> ActionDefinition {
440    ActionDefinition {
441        name: "deliver_nsq_messages".to_string(),
442        description: "Deliver messages to this subscriber. NetGet gives each one its message id \
443                      and timestamp, and sends no more than the client's RDY count allows; the \
444                      rest wait in order until the client finishes one. Only a subscribed \
445                      connection can receive messages."
446            .to_string(),
447        parameters: vec![Parameter {
448            name: "messages".to_string(),
449            type_hint: "array".to_string(),
450            description: "List of {body, attempts}: body is the message text, attempts \
451                          (optional, default 1) how many times it has been delivered"
452                .to_string(),
453            required: true,
454        }],
455        example: json!({
456            "type": "deliver_nsq_messages",
457            "messages": [{"body": "<message body>"}]
458        }),
459        log_template: Some(
460            LogTemplate::new()
461                .with_info("-> NSQ deliver messages")
462                .with_debug("NSQ deliver_nsq_messages"),
463        ),
464    }
465}
466
467fn close_connection_action() -> ActionDefinition {
468    ActionDefinition {
469        name: "close_connection".to_string(),
470        description: "Close the NSQ connection after any reply".to_string(),
471        parameters: vec![],
472        example: json!({"type": "close_connection"}),
473        log_template: Some(
474            LogTemplate::new()
475                .with_info("NSQ connection closed")
476                .with_debug("NSQ close_connection"),
477        ),
478    }
479}
480
481fn param(name: &str, type_hint: &str, description: &str) -> Parameter {
482    Parameter {
483        name: name.to_string(),
484        type_hint: type_hint.to_string(),
485        description: description.to_string(),
486        required: true,
487    }
488}
489
490fn answer_with_param() -> Parameter {
491    param(
492        "answer_with",
493        "string",
494        "Which answer this request takes, and where its content comes from",
495    )
496}
497
498/// The `answer_with` field of `nsq_publish`.
499pub fn publish_answer_with(command: &str, topic: &str, count: usize) -> String {
500    let refusal = match command {
501        "MPUB" => "E_MPUB_FAILED",
502        "DPUB" => "E_DPUB_FAILED",
503        _ => "E_PUB_FAILED",
504    };
505    format!(
506        "exactly one action: send_nsq_ok to accept the {count} message(s) published to topic \
507         '{topic}', or send_nsq_error with code {refusal} only if your instructions say this \
508         topic refuses publishes"
509    )
510}
511
512/// The `answer_with` field of `nsq_subscribe`.
513pub fn subscribe_answer_with(topic: &str, channel: &str) -> String {
514    format!(
515        "exactly one action: send_nsq_ok to accept the subscription to topic '{topic}' on \
516         channel '{channel}', or send_nsq_error (E_BAD_TOPIC, E_BAD_CHANNEL or E_UNAUTHORIZED) \
517         only if your instructions say it is refused. No messages yet: the client has not sent \
518         RDY"
519    )
520}
521
522/// The `answer_with` field of `nsq_ready`, `nsq_finish` and `nsq_requeue` for a delivery.
523pub fn deliver_answer_with(topic: &str, can_deliver: u64, pending: usize) -> String {
524    let queued = if pending > 0 {
525        format!(" {pending} message(s) you gave earlier are already queued and will follow;")
526    } else {
527        String::new()
528    };
529    format!(
530        "deliver_nsq_messages with at most {can_deliver} message(s) your instructions say are \
531         waiting on topic '{topic}' that you have not delivered before, each body word for \
532         word as your instructions give it;{queued} no action at all when none are waiting"
533    )
534}
535
536/// The `answer_with` field of `nsq_requeue`.
537pub fn requeue_answer_with(attempts: u16) -> String {
538    format!(
539        "deliver_nsq_messages with this message's body again and attempts {} if your \
540         instructions say a requeued message is retried; no action at all to drop it",
541        attempts.saturating_add(1)
542    )
543}
544
545fn deliver_actions() -> Vec<ActionDefinition> {
546    vec![deliver_action(), error_action(), close_connection_action()]
547}
548
549/// `PUB`, `MPUB` and `DPUB`.
550pub static NSQ_PUBLISH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
551    EventType::new(
552        "nsq_publish",
553        "A client published messages to a topic. Accept them with send_nsq_ok or refuse them \
554         with send_nsq_error; exactly one of the two. answer_with says which fits.",
555        json!({"type": "send_nsq_ok"}),
556    )
557    .with_parameters(vec![
558        param("command", "string", "PUB, MPUB or DPUB").with_choices(["PUB", "MPUB", "DPUB"]),
559        param("topic", "string", "The topic published to"),
560        param(
561            "messages",
562            "array",
563            "The message bodies as text (invalid UTF-8 replaced), the first 20, each cut at \
564             1000 characters",
565        ),
566        param(
567            "message_count",
568            "number",
569            "How many messages were published",
570        ),
571        param("total_bytes", "number", "The bodies' total size in bytes"),
572        answer_with_param(),
573    ])
574    .with_log_template(
575        LogTemplate::new()
576            .with_info("NSQ {command} {topic} ({message_count} message(s))")
577            .with_debug("NSQ nsq_publish: topic={topic} bytes={total_bytes}"),
578    )
579    .with_actions(vec![
580        ok_action(),
581        error_action(),
582        deliver_action(),
583        close_connection_action(),
584    ])
585    .with_alternative_example(
586        json!({"type": "send_nsq_error", "code": "E_PUB_FAILED", "message": "<why>"}),
587    )
588});
589
590/// `SUB`.
591pub static NSQ_SUBSCRIBE_EVENT: LazyLock<EventType> =
592    LazyLock::new(|| {
593        EventType::new(
594            "nsq_subscribe",
595            "A client subscribed to a topic's channel. Accept with send_nsq_ok or refuse with \
596         send_nsq_error; exactly one of the two. Messages cannot be delivered yet.",
597            json!({"type": "send_nsq_ok"}),
598        )
599        .with_parameters(vec![
600        param("topic", "string", "The topic the client subscribes to"),
601        param(
602            "channel",
603            "string",
604            "The channel on that topic; each channel receives its own copy of the topic's messages",
605        ),
606        param("ephemeral", "boolean", "Whether the channel name ends in #ephemeral"),
607        param("client_id", "string", "The client_id from IDENTIFY (may be empty)"),
608        param("user_agent", "string", "The user_agent from IDENTIFY (may be empty)"),
609        answer_with_param(),
610    ])
611        .with_log_template(
612            LogTemplate::new()
613                .with_info("NSQ SUB {topic} {channel}")
614                .with_debug("NSQ nsq_subscribe: topic={topic} channel={channel}"),
615        )
616        .with_actions(vec![ok_action(), error_action(), close_connection_action()])
617        .with_alternative_example(
618            json!({"type": "send_nsq_error", "code": "E_UNAUTHORIZED", "message": "<why>"}),
619        )
620    });
621
622fn flow_parameters() -> Vec<Parameter> {
623    vec![
624        param("topic", "string", "The subscribed topic"),
625        param("channel", "string", "The subscribed channel"),
626        param(
627            "ready",
628            "number",
629            "The client's RDY count: how many messages it takes at once",
630        ),
631        param(
632            "in_flight",
633            "number",
634            "Messages delivered and not yet finished or requeued",
635        ),
636        param(
637            "pending",
638            "number",
639            "Messages you already gave that wait for RDY capacity; NetGet sends them first",
640        ),
641        param(
642            "can_deliver",
643            "number",
644            "How many more messages the client can receive right now",
645        ),
646    ]
647}
648
649/// `RDY` with room for at least one message.
650pub static NSQ_READY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
651    let mut parameters = vec![param(
652        "count",
653        "number",
654        "The RDY count the client just sent",
655    )];
656    parameters.extend(flow_parameters());
657    parameters.push(answer_with_param());
658    EventType::new(
659        "nsq_ready",
660        "A subscriber is ready for messages (RDY). Deliver the messages your instructions say \
661         are waiting with deliver_nsq_messages, or answer with no action when there are none.",
662        json!({"type": "deliver_nsq_messages", "messages": [{"body": "<message body>"}]}),
663    )
664    .with_parameters(parameters)
665    .with_log_template(
666        LogTemplate::new()
667            .with_info("NSQ RDY {count} on {topic}/{channel}")
668            .with_debug("NSQ nsq_ready: in_flight={in_flight} can_deliver={can_deliver}"),
669    )
670    .with_actions(deliver_actions())
671});
672
673/// `FIN` of a message in flight.
674pub static NSQ_FINISH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
675    let mut parameters = vec![param(
676        "message_id",
677        "string",
678        "The id of the message the client finished",
679    )];
680    parameters.extend(flow_parameters());
681    parameters.push(answer_with_param());
682    EventType::new(
683        "nsq_finish",
684        "A subscriber finished (acknowledged) a message. Deliver more with \
685         deliver_nsq_messages only if your instructions say more are waiting; otherwise \
686         answer with no action.",
687        json!({"type": "deliver_nsq_messages", "messages": [{"body": "<message body>"}]}),
688    )
689    .with_parameters(parameters)
690    .with_log_template(
691        LogTemplate::new()
692            .with_info("NSQ FIN {message_id}")
693            .with_debug("NSQ nsq_finish: in_flight={in_flight} pending={pending}"),
694    )
695    .with_actions(deliver_actions())
696});
697
698/// `REQ` of a message in flight.
699pub static NSQ_REQUEUE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
700    let mut parameters = vec![
701        param(
702            "message_id",
703            "string",
704            "The id of the message the client requeued",
705        ),
706        param(
707            "timeout_ms",
708            "number",
709            "How long the client asked the message to wait before redelivery",
710        ),
711        param(
712            "attempts",
713            "number",
714            "How many times the message has been delivered",
715        ),
716        param("body", "string", "The requeued message's body"),
717    ];
718    parameters.extend(flow_parameters());
719    parameters.push(answer_with_param());
720    EventType::new(
721        "nsq_requeue",
722        "A subscriber requeued a message (it could not process it now). Redeliver it with \
723         deliver_nsq_messages and attempts one higher if your instructions say to retry; \
724         otherwise answer with no action.",
725        json!({"type": "deliver_nsq_messages", "messages": [{"body": "<message body>", "attempts": 2}]}),
726    )
727    .with_parameters(parameters)
728    .with_log_template(
729        LogTemplate::new()
730            .with_info("NSQ REQ {message_id} ({timeout_ms} ms)")
731            .with_debug("NSQ nsq_requeue: attempts={attempts}"),
732    )
733    .with_actions(deliver_actions())
734});
netget/server/nsq/actions.rs

netget/server/nsq/
actions.rs
