netget/server/stomp/
actions.rs

1//! STOMP 1.2 protocol actions.
2//!
3//! What the model decides here is *content*: whether to admit a connection, what a MESSAGE
4//! carries, whether to refuse. Framing, header escaping, the NUL terminator, the `receipt`
5//! handshake and the heart-beat negotiation are all done in Rust (`frame.rs` and `mod.rs`) —
6//! they are mechanical, and a model that gets any of them wrong produces a stream no real
7//! client can recover from.
8
9use crate::llm::actions::{
10    protocol_trait::{ActionResult, Protocol, Server},
11    ActionDefinition, Parameter,
12};
13use crate::protocol::log_template::LogTemplate;
14use crate::protocol::EventType;
15use crate::server::stomp::frame::{error_frame, receipt_frame, StompFrame};
16use crate::state::app_state::AppState;
17use anyhow::{Context, Result};
18use serde_json::json;
19use std::sync::LazyLock;
20
21/// The version this server speaks. Only 1.2 — `mod.rs` refuses a `CONNECT` whose
22/// `accept-version` does not include it, rather than silently answering a dialect it does not
23/// implement.
24pub const STOMP_VERSION: &str = "1.2";
25
26/// Heart-beating is negotiated off, unconditionally.
27///
28/// The server implements no heart-beat timer, so any other value would be a promise it does
29/// not keep — and a client that believes it will hear from us every N ms tears the connection
30/// down when it does not. `send_stomp_connected` deliberately has **no** heart-beat parameter
31/// for the same reason: the model cannot advertise what the code does not do.
32pub const STOMP_HEARTBEAT: &str = "0,0";
33
34pub struct StompProtocol;
35
36impl StompProtocol {
37    pub fn new() -> Self {
38        Self
39    }
40}
41
42impl Default for StompProtocol {
43    fn default() -> Self {
44        Self::new()
45    }
46}
47
48impl Protocol for StompProtocol {
49    fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
50        // STOMP is purely reactive: every frame this server writes answers one the client
51        // sent. Nothing here can be issued out of band.
52        Vec::new()
53    }
54
55    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
56        vec![
57            send_stomp_connected_action(),
58            send_stomp_message_action(),
59            send_stomp_receipt_action(),
60            send_stomp_error_action(),
61            close_connection_action(),
62        ]
63    }
64
65    fn protocol_name(&self) -> &'static str {
66        "STOMP"
67    }
68
69    fn get_event_types(&self) -> Vec<EventType> {
70        get_stomp_event_types()
71    }
72
73    fn stack_name(&self) -> &'static str {
74        "ETH>IP>TCP>STOMP"
75    }
76
77    fn keywords(&self) -> Vec<&'static str> {
78        vec!["stomp"]
79    }
80
81    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
82        use crate::protocol::metadata::{
83            DevelopmentState, PrivilegeRequirement, ProtocolMetadataV2,
84        };
85
86        ProtocolMetadataV2::builder()
87            .state(DevelopmentState::Beta)
88            .privilege_requirement(PrivilegeRequirement::None)
89            .well_known_port(61613)
90            .implementation(
91                "Hand-rolled STOMP 1.2 codec over tokio TCP (src/server/stomp/frame.rs). No \
92                 third-party STOMP crate on the server side: header escaping, the \
93                 content-length/NUL body rule and the receipt handshake are ~300 lines and a \
94                 dependency would not have removed the part that matters.",
95            )
96            .llm_control(
97                "Whether to admit a CONNECT, the contents of MESSAGE frames, and whether to \
98                 refuse with ERROR. Framing, header escaping, receipts and heart-beat \
99                 negotiation are deterministic Rust.",
100            )
101            .e2e_testing(
102                "Driven by the async-stomp 0.6.3 client (tests/server/stomp/e2e_test.rs), which \
103                 is not #[ignore]d and cannot skip - it is a compiled-in crate dependency, so \
104                 there is nothing to detect as missing. Connector::connect() completes the \
105                 handshake and async-stomp's own parser decodes every later frame. \
106                 raw_socket_test.rs covers what a client library cannot express (a NUL body, a \
107                 frame before the handshake, a refused accept-version, broken framing) and \
108                 codec_test.rs pins the codec against spec byte literals.",
109            )
110            .notes(
111                "Beta on the evidence of a real third-party client: async-stomp 0.6.3 completes \
112                 CONNECT -> CONNECTED -> SUBSCRIBE -> MESSAGE -> SEND -> MESSAGE -> DISCONNECT \
113                 -> RECEIPT -> close against this server, decoding each frame with its own \
114                 parser, and separately decodes a refusal as an ERROR frame. A negative control \
115                 was run: dropping the CONNECTED 'version' header makes async-stomp's handshake \
116                 fail and the test fail, so it is not passing vacuously. NOT IMPLEMENTED, and \
117                 none of it is hidden from the peer: heart-beating (always negotiated 0,0, \
118                 which is a legal configuration rather than a silent omission), transactions \
119                 (BEGIN/COMMIT/ABORT are acknowledged and otherwise ignored - holding messages \
120                 until COMMIT would be storage, which protocols may not implement; the \
121                 transaction header still reaches the model on the SEND event), subscription \
122                 bookkeeping, message redelivery, STOMP 1.0/1.1 (refused with an ERROR naming \
123                 version:1.2) and TLS. UNPROVEN: interop with brokers' own clients \
124                 (ActiveMQ/RabbitMQ STOMP), and concurrent sessions.",
125            )
126            .max_inbound_bytes(crate::server::stomp::frame::MAX_FRAME_BYTES)
127            .build()
128    }
129
130    fn description(&self) -> &'static str {
131        "STOMP 1.2 message broker server"
132    }
133
134    fn example_prompt(&self) -> &'static str {
135        "STOMP broker on port 61613 - accept any login and deliver a greeting message to \
136         whoever subscribes to /queue/test"
137    }
138
139    fn group_name(&self) -> &'static str {
140        "Application"
141    }
142
143    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
144        use crate::llm::actions::StartupExamples;
145        StartupExamples::new(
146            json!({
147                "type": "open_server",
148                "port": 61613,
149                "base_stack": "stomp",
150                "instruction": "STOMP broker: accept every CONNECT, and when a client subscribes to a destination send it one MESSAGE describing that destination"
151            }),
152            json!({
153                "type": "open_server",
154                "port": 61613,
155                "base_stack": "stomp",
156                "event_handlers": [{
157                    "event_pattern": "stomp_connect",
158                    "handler": {
159                        "type": "script",
160                        "language": "python",
161                        "code": "# Admit anyone, naming the session after the login they offered\nlogin = event.get('login') or 'anonymous'\nrespond([{'type': 'send_stomp_connected', 'session': 'session-' + login, 'server': 'netget/stomp'}])"
162                    }
163                }]
164            }),
165            json!({
166                "type": "open_server",
167                "port": 61613,
168                "base_stack": "stomp",
169                "event_handlers": [{
170                    "event_pattern": "stomp_connect",
171                    "handler": {
172                        "type": "static",
173                        "actions": [{
174                            "type": "send_stomp_connected",
175                            "version": "1.2",
176                            "session": "session-1",
177                            "server": "netget/stomp"
178                        }]
179                    }
180                }]
181            }),
182        )
183    }
184}
185
186impl Server for StompProtocol {
187    fn spawn(
188        &self,
189        ctx: crate::protocol::SpawnContext,
190    ) -> std::pin::Pin<
191        Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
192    > {
193        Box::pin(async move {
194            use crate::server::stomp::StompServer;
195            StompServer::spawn_with_llm_actions(
196                ctx.legacy_listen_addr(),
197                ctx.llm_client,
198                ctx.state,
199                ctx.status_tx,
200                ctx.server_id,
201            )
202            .await
203        })
204    }
205
206    fn execute_action(&self, action: serde_json::Value) -> Result<ActionResult> {
207        let action_type = action
208            .get("type")
209            .and_then(|v| v.as_str())
210            .context("Missing 'type' field in action")?;
211
212        match action_type {
213            "send_stomp_connected" => execute_send_connected(&action),
214            "send_stomp_message" => execute_send_message(&action),
215            "send_stomp_receipt" => execute_send_receipt(&action),
216            "send_stomp_error" => execute_send_error(&action),
217            "close_connection" => Ok(ActionResult::CloseConnection),
218            other => Err(anyhow::anyhow!("Unknown STOMP action: {}", other)),
219        }
220    }
221}
222
223// === executors ===
224
225fn execute_send_connected(action: &serde_json::Value) -> Result<ActionResult> {
226    let version = action
227        .get("version")
228        .and_then(|v| v.as_str())
229        .unwrap_or(STOMP_VERSION);
230    if version != STOMP_VERSION {
231        return Err(anyhow::anyhow!(
232            "This server implements STOMP {STOMP_VERSION} only; 'version' must be \
233             \"{STOMP_VERSION}\" (got {version:?})"
234        ));
235    }
236
237    let mut headers = vec![("version".to_string(), version.to_string())];
238    if let Some(session) = action.get("session").and_then(|v| v.as_str()) {
239        reject_unescapable("session", session)?;
240        headers.push(("session".to_string(), session.to_string()));
241    }
242    if let Some(server) = action.get("server").and_then(|v| v.as_str()) {
243        reject_unescapable("server", server)?;
244        headers.push(("server".to_string(), server.to_string()));
245    }
246    // Never taken from the action: see STOMP_HEARTBEAT.
247    headers.push(("heart-beat".to_string(), STOMP_HEARTBEAT.to_string()));
248
249    Ok(ActionResult::Output(
250        StompFrame::new("CONNECTED", headers, Vec::new()).encode(),
251    ))
252}
253
254/// Refuse a value that would forge a header on a `CONNECTED` frame.
255///
256/// This is the one place in this protocol where the model's string reaches the wire without
257/// the escaper in front of it: STOMP 1.2 exempts `CONNECT`/`STOMP`/`CONNECTED` from escaping,
258/// so `encode()` writes their header values raw. A `session` containing a newline injects an
259/// arbitrary extra header; `\n\n` ends the header block and starts a body. The protocol's own
260/// startup example builds the session out of peer input (`'session-' + login`), so this is
261/// reachable from a script handler as readily as from the model.
262///
263/// It also keeps the documented interop honest: `async-stomp` unescapes `CONNECTED` headers
264/// unconditionally, so a value with a backslash in it would come back changed.
265fn reject_unescapable(field: &str, value: &str) -> Result<()> {
266    if !crate::server::stomp::frame::is_safe_unescaped_header(value) {
267        return Err(anyhow::anyhow!(
268            "'{field}' may not contain a newline, carriage return, colon or NUL: CONNECTED \
269             headers are exempt from STOMP 1.2 escaping, so such a value would forge a header \
270             rather than appear in this one"
271        ));
272    }
273    Ok(())
274}
275
276fn execute_send_message(action: &serde_json::Value) -> Result<ActionResult> {
277    let destination = required_str(action, "destination")?;
278    let subscription = required_str(action, "subscription")?;
279    let message_id = required_str(action, "message_id")?;
280    let body = decode_body(action)?;
281
282    let mut headers = vec![
283        ("destination".to_string(), destination),
284        ("subscription".to_string(), subscription),
285        ("message-id".to_string(), message_id),
286    ];
287    if let Some(content_type) = action.get("content_type").and_then(|v| v.as_str()) {
288        headers.push(("content-type".to_string(), content_type.to_string()));
289    }
290    headers.extend(extra_headers(action)?);
291
292    Ok(ActionResult::Output(
293        StompFrame::new("MESSAGE", headers, body).encode(),
294    ))
295}
296
297fn execute_send_receipt(action: &serde_json::Value) -> Result<ActionResult> {
298    let receipt_id = required_str(action, "receipt_id")?;
299    Ok(ActionResult::Output(receipt_frame(&receipt_id)))
300}
301
302fn execute_send_error(action: &serde_json::Value) -> Result<ActionResult> {
303    let message = required_str(action, "message")?;
304    let body = action
305        .get("body")
306        .and_then(|v| v.as_str())
307        .unwrap_or("")
308        .to_string();
309    Ok(ActionResult::Output(error_frame(&message, &body)))
310}
311
312fn required_str(action: &serde_json::Value, key: &str) -> Result<String> {
313    action
314        .get(key)
315        .and_then(|v| v.as_str())
316        .map(|s| s.to_string())
317        .with_context(|| format!("Missing or non-string '{key}' parameter"))
318}
319
320/// Decode `body` according to the **explicit** `encoding` field.
321///
322/// There is no sniffing, deliberately: `"48656c6c6f"` is simultaneously valid text and valid
323/// hex, and only the sender knows which it means. This is the same contract the inbound
324/// `stomp_send` event uses (`body` plus `body_encoding`), so echoing a received body straight
325/// back reproduces the exact bytes.
326fn decode_body(action: &serde_json::Value) -> Result<Vec<u8>> {
327    let body = action.get("body").and_then(|v| v.as_str()).unwrap_or("");
328    match action
329        .get("encoding")
330        .and_then(|v| v.as_str())
331        .unwrap_or("utf8")
332    {
333        "utf8" => Ok(body.as_bytes().to_vec()),
334        "hex" => {
335            let cleaned: String = body
336                .trim()
337                .trim_start_matches("0x")
338                .chars()
339                .filter(|c| !c.is_whitespace() && *c != ':')
340                .collect();
341            hex::decode(&cleaned).map_err(|e| {
342                anyhow::anyhow!("'body' is not valid hex although encoding is \"hex\": {e}")
343            })
344        }
345        other => Err(anyhow::anyhow!(
346            "Unknown 'encoding' {other:?}; valid values are \"utf8\" (default) and \"hex\""
347        )),
348    }
349}
350
351/// Model-supplied extra headers.
352///
353/// `content-length` is dropped: it is computed from the body the model actually sent, and a
354/// wrong one supplied here would truncate the frame or make the peer wait for bytes that never
355/// come. Nested values are refused rather than stringified, because there is no
356/// representation of a JSON object in a STOMP header that a client could read back.
357fn extra_headers(action: &serde_json::Value) -> Result<Vec<(String, String)>> {
358    let Some(map) = action.get("headers").and_then(|v| v.as_object()) else {
359        return Ok(Vec::new());
360    };
361    let mut out = Vec::new();
362    for (k, v) in map {
363        if k == "content-length" {
364            continue;
365        }
366        let value = match v {
367            serde_json::Value::String(s) => s.clone(),
368            serde_json::Value::Number(n) => n.to_string(),
369            serde_json::Value::Bool(b) => b.to_string(),
370            serde_json::Value::Null => continue,
371            _ => {
372                return Err(anyhow::anyhow!(
373                    "header {k:?} must be a string, number or boolean; a STOMP header cannot \
374                     carry a nested object or array"
375                ))
376            }
377        };
378        out.push((k.clone(), value));
379    }
380    Ok(out)
381}
382
383// === action definitions ===
384
385fn send_stomp_connected_action() -> ActionDefinition {
386    ActionDefinition {
387        name: "send_stomp_connected".to_string(),
388        description: format!(
389            "Accept the client's CONNECT and open the session. This or send_stomp_error is the \
390             only valid answer to stomp_connect - STOMP has no third option, and a client that \
391             gets neither blocks. Heart-beating is always negotiated off ({STOMP_HEARTBEAT}) \
392             because this server implements no heart-beat timer, so there is no parameter for it."
393        ),
394        parameters: vec![
395            Parameter {
396                name: "version".to_string(),
397                type_hint: "string".to_string(),
398                description: format!(
399                    "Negotiated protocol version. Only \"{STOMP_VERSION}\" is implemented; \
400                     omit to use it."
401                ),
402                required: false,
403            },
404            Parameter {
405                name: "session".to_string(),
406                type_hint: "string".to_string(),
407                description: "Session identifier shown to the client".to_string(),
408                required: false,
409            },
410            Parameter {
411                name: "server".to_string(),
412                type_hint: "string".to_string(),
413                description: "Server name/version string, e.g. \"netget/stomp\"".to_string(),
414                required: false,
415            },
416        ],
417        example: json!({
418            "type": "send_stomp_connected",
419            "version": "1.2",
420            "session": "session-1",
421            "server": "netget/stomp"
422        }),
423        log_template: Some(
424            LogTemplate::new()
425                .with_info("-> STOMP CONNECTED (session {session})")
426                .with_debug("STOMP send_stomp_connected: session={session} server={server}"),
427        ),
428    }
429}
430
431fn send_stomp_message_action() -> ActionDefinition {
432    ActionDefinition {
433        name: "send_stomp_message".to_string(),
434        description: "Deliver a MESSAGE frame to the client. `subscription` must be the id the \
435                      client used in its SUBSCRIBE, or the client will discard the frame; \
436                      `message_id` is yours to choose and must be unique within the session."
437            .to_string(),
438        parameters: vec![
439            Parameter {
440                name: "destination".to_string(),
441                type_hint: "string".to_string(),
442                description: "Destination the message belongs to, e.g. \"/queue/test\"".to_string(),
443                required: true,
444            },
445            Parameter {
446                name: "subscription".to_string(),
447                type_hint: "string".to_string(),
448                description: "The subscription id from the client's SUBSCRIBE frame".to_string(),
449                required: true,
450            },
451            Parameter {
452                name: "message_id".to_string(),
453                type_hint: "string".to_string(),
454                description: "Unique identifier for this message".to_string(),
455                required: true,
456            },
457            Parameter {
458                name: "body".to_string(),
459                type_hint: "string".to_string(),
460                description: "Message body, interpreted according to `encoding`".to_string(),
461                required: false,
462            },
463            Parameter {
464                name: "encoding".to_string(),
465                type_hint: "string".to_string(),
466                description: "How to read `body`: \"utf8\" (default) sends the characters as \
467                              they are; \"hex\" decodes hex digits into bytes. There is no \
468                              sniffing - say which you mean."
469                    .to_string(),
470                required: false,
471            },
472            Parameter {
473                name: "content_type".to_string(),
474                type_hint: "string".to_string(),
475                description: "MIME type of the body, e.g. \"text/plain\"".to_string(),
476                required: false,
477            },
478            Parameter {
479                name: "headers".to_string(),
480                type_hint: "object".to_string(),
481                description: "Additional headers as a flat string-to-string map. \
482                              `content-length` is ignored - it is computed from the body."
483                    .to_string(),
484                required: false,
485            },
486        ],
487        example: json!({
488            "type": "send_stomp_message",
489            "destination": "/queue/test",
490            "subscription": "sub-0",
491            "message_id": "msg-1",
492            "content_type": "text/plain",
493            "body": "hello from netget",
494            "encoding": "utf8"
495        }),
496        log_template: Some(
497            LogTemplate::new()
498                .with_info("-> STOMP MESSAGE {destination} (sub {subscription})")
499                .with_debug(
500                    "STOMP send_stomp_message: destination={destination} \
501                     subscription={subscription} message_id={message_id}",
502                ),
503        ),
504    }
505}
506
507fn send_stomp_receipt_action() -> ActionDefinition {
508    ActionDefinition {
509        name: "send_stomp_receipt".to_string(),
510        description: "Send a RECEIPT frame. You rarely need this: any client frame carrying a \
511                      `receipt` header is answered with a RECEIPT automatically once the frame \
512                      has been processed. Use it only to acknowledge something out of band."
513            .to_string(),
514        parameters: vec![Parameter {
515            name: "receipt_id".to_string(),
516            type_hint: "string".to_string(),
517            description: "Value of the client's `receipt` header".to_string(),
518            required: true,
519        }],
520        example: json!({"type": "send_stomp_receipt", "receipt_id": "receipt-1"}),
521        log_template: Some(
522            LogTemplate::new()
523                .with_info("-> STOMP RECEIPT {receipt_id}")
524                .with_debug("STOMP send_stomp_receipt: receipt_id={receipt_id}"),
525        ),
526    }
527}
528
529fn send_stomp_error_action() -> ActionDefinition {
530    ActionDefinition {
531        name: "send_stomp_error".to_string(),
532        description: "Refuse with an ERROR frame. The STOMP spec requires the connection to be \
533                      closed afterwards, and this server does so automatically - so this is a \
534                      final answer, not something to follow with more frames."
535            .to_string(),
536        parameters: vec![
537            Parameter {
538                name: "message".to_string(),
539                type_hint: "string".to_string(),
540                description: "Short reason, becomes the `message` header".to_string(),
541                required: true,
542            },
543            Parameter {
544                name: "body".to_string(),
545                type_hint: "string".to_string(),
546                description: "Longer explanation, becomes the frame body".to_string(),
547                required: false,
548            },
549        ],
550        example: json!({
551            "type": "send_stomp_error",
552            "message": "authentication failed",
553            "body": "The login/passcode pair was not recognised."
554        }),
555        log_template: Some(
556            LogTemplate::new()
557                .with_info("-> STOMP ERROR: {message}")
558                .with_debug("STOMP send_stomp_error: {message}"),
559        ),
560    }
561}
562
563fn close_connection_action() -> ActionDefinition {
564    ActionDefinition {
565        name: "close_connection".to_string(),
566        description: "Close the STOMP connection without saying anything further".to_string(),
567        parameters: vec![],
568        example: json!({"type": "close_connection"}),
569        log_template: Some(
570            LogTemplate::new()
571                .with_info("STOMP connection closed")
572                .with_debug("STOMP close_connection"),
573        ),
574    }
575}
576
577// === event types ===
578
579/// Actions that make sense once a session is open.
580fn session_actions() -> Vec<ActionDefinition> {
581    vec![
582        send_stomp_message_action(),
583        send_stomp_receipt_action(),
584        send_stomp_error_action(),
585        close_connection_action(),
586    ]
587}
588
589/// Client sent `CONNECT` (or its `STOMP` synonym).
590pub static STOMP_CONNECT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
591    EventType::new(
592        "stomp_connect",
593        "Client opened a STOMP session with a CONNECT or STOMP frame. Answer with \
594         send_stomp_connected to admit it or send_stomp_error to refuse; the client blocks \
595         until one of the two arrives.",
596        json!({
597            "type": "send_stomp_connected",
598            "version": "1.2",
599            "session": "session-1",
600            "server": "netget/stomp"
601        }),
602    )
603    .with_parameters(vec![
604        Parameter {
605            name: "accept_version".to_string(),
606            type_hint: "string".to_string(),
607            description: "Comma-separated versions the client accepts".to_string(),
608            required: false,
609        },
610        Parameter {
611            name: "host".to_string(),
612            type_hint: "string".to_string(),
613            description: "Virtual host the client asked for".to_string(),
614            required: false,
615        },
616        Parameter {
617            name: "login".to_string(),
618            type_hint: "string".to_string(),
619            description: "Login name offered by the client, if any".to_string(),
620            required: false,
621        },
622        Parameter {
623            name: "passcode".to_string(),
624            type_hint: "string".to_string(),
625            description: "Passcode offered by the client, if any".to_string(),
626            required: false,
627        },
628        Parameter {
629            name: "heart_beat".to_string(),
630            type_hint: "string".to_string(),
631            description: "The client's requested heart-beat pair. Informational only: this \
632                          server always negotiates 0,0."
633                .to_string(),
634            required: false,
635        },
636    ])
637    .with_log_template(
638        LogTemplate::new()
639            .with_info("STOMP CONNECT host={host} login={login}")
640            .with_debug("STOMP connect: accept_version={accept_version} host={host} login={login}")
641            .with_trace("STOMP connect: {json_pretty(.)}"),
642    )
643    .with_actions(vec![
644        send_stomp_connected_action(),
645        send_stomp_error_action(),
646        close_connection_action(),
647    ])
648    .with_alternative_example(json!({
649        "type": "send_stomp_error",
650        "message": "authentication failed",
651        "body": "The login/passcode pair was not recognised."
652    }))
653});
654
655/// Client published a frame with `SEND`.
656pub static STOMP_SEND_EVENT: LazyLock<EventType> = LazyLock::new(|| {
657    EventType::new(
658        "stomp_send",
659        "Client published a message to a destination. A broker normally says nothing in reply, \
660         so answering with no action is legitimate; deliver it onward with send_stomp_message \
661         if a subscriber should see it.",
662        json!({
663            "type": "send_stomp_message",
664            "destination": "/queue/test",
665            "subscription": "sub-0",
666            "message_id": "msg-1",
667            "body": "hello from netget",
668            "encoding": "utf8"
669        }),
670    )
671    .with_parameters(vec![
672        Parameter {
673            name: "destination".to_string(),
674            type_hint: "string".to_string(),
675            description: "Destination the client published to".to_string(),
676            required: true,
677        },
678        Parameter {
679            name: "body".to_string(),
680            type_hint: "string".to_string(),
681            description: "The message body, read according to body_encoding".to_string(),
682            required: false,
683        },
684        Parameter {
685            name: "body_encoding".to_string(),
686            type_hint: "string".to_string(),
687            description: "\"utf8\" when the body was printable text, \"hex\" when it was not. \
688                          Pass it through to send_stomp_message's `encoding` to echo the exact \
689                          bytes back."
690                .to_string(),
691            required: true,
692        },
693        Parameter {
694            name: "headers".to_string(),
695            type_hint: "object".to_string(),
696            description: "Every other header on the frame, including content-type and any \
697                          transaction id"
698                .to_string(),
699            required: false,
700        },
701    ])
702    .with_log_template(
703        LogTemplate::new()
704            .with_info("STOMP SEND {destination}")
705            .with_debug("STOMP send: destination={destination} encoding={body_encoding}")
706            .with_trace("STOMP send: {json_pretty(.)}"),
707    )
708    .with_actions(session_actions())
709});
710
711/// Client subscribed to a destination.
712pub static STOMP_SUBSCRIBE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
713    EventType::new(
714        "stomp_subscribe",
715        "Client subscribed to a destination. Send it messages with send_stomp_message, quoting \
716         this subscription id.",
717        json!({
718            "type": "send_stomp_message",
719            "destination": "/queue/test",
720            "subscription": "sub-0",
721            "message_id": "msg-1",
722            "body": "welcome",
723            "encoding": "utf8"
724        }),
725    )
726    .with_parameters(vec![
727        Parameter {
728            name: "destination".to_string(),
729            type_hint: "string".to_string(),
730            description: "Destination subscribed to".to_string(),
731            required: true,
732        },
733        Parameter {
734            name: "id".to_string(),
735            type_hint: "string".to_string(),
736            description: "Subscription id chosen by the client; quote it in send_stomp_message"
737                .to_string(),
738            required: true,
739        },
740        Parameter {
741            name: "ack_mode".to_string(),
742            type_hint: "string".to_string(),
743            description: "\"auto\" (default), \"client\" or \"client-individual\"".to_string(),
744            required: false,
745        },
746    ])
747    .with_log_template(
748        LogTemplate::new()
749            .with_info("STOMP SUBSCRIBE {destination} (id {id})")
750            .with_debug("STOMP subscribe: destination={destination} id={id} ack={ack_mode}")
751            .with_trace("STOMP subscribe: {json_pretty(.)}"),
752    )
753    .with_actions(session_actions())
754});
755
756/// Client cancelled a subscription.
757pub static STOMP_UNSUBSCRIBE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
758    EventType::new(
759        "stomp_unsubscribe",
760        "Client cancelled a subscription. Answering with no action is the normal case - any \
761         receipt the frame asked for is sent for you. Use send_stomp_message only if a \
762         message still has to go out on another subscription.",
763        json!({
764            "type": "send_stomp_message",
765            "destination": "/queue/test",
766            "subscription": "sub-1",
767            "message_id": "msg-9",
768            "body": "sent on a subscription that is still open",
769            "encoding": "utf8"
770        }),
771    )
772    .with_parameters(vec![Parameter {
773        name: "id".to_string(),
774        type_hint: "string".to_string(),
775        description: "Subscription id being cancelled".to_string(),
776        required: true,
777    }])
778    .with_log_template(
779        LogTemplate::new()
780            .with_info("STOMP UNSUBSCRIBE id={id}")
781            .with_debug("STOMP unsubscribe: id={id}")
782            .with_trace("STOMP unsubscribe: {json_pretty(.)}"),
783    )
784    .with_actions(session_actions())
785});
786
787/// Client acknowledged a message.
788pub static STOMP_ACK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
789    EventType::new(
790        "stomp_ack",
791        "Client acknowledged a message it had been sent. Answering with no action is \
792         legitimate; the usual reason to answer is to send the next message now that this one \
793         is settled. Any receipt the frame asked for is sent for you.",
794        json!({
795            "type": "send_stomp_message",
796            "destination": "/queue/test",
797            "subscription": "sub-0",
798            "message_id": "msg-2",
799            "body": "the next message",
800            "encoding": "utf8"
801        }),
802    )
803    .with_parameters(vec![Parameter {
804        name: "id".to_string(),
805        type_hint: "string".to_string(),
806        description: "The ack id, which is the message-id of the MESSAGE being acknowledged"
807            .to_string(),
808        required: true,
809    }])
810    .with_log_template(
811        LogTemplate::new()
812            .with_info("STOMP ACK id={id}")
813            .with_debug("STOMP ack: id={id}")
814            .with_trace("STOMP ack: {json_pretty(.)}"),
815    )
816    .with_actions(session_actions())
817});
818
819/// Client rejected a message.
820pub static STOMP_NACK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
821    EventType::new(
822        "stomp_nack",
823        "Client rejected a message it had been sent. This server does not redeliver on its own \
824         - resend it with send_stomp_message if that is what you want, or answer with no \
825         action to drop it. Any receipt the frame asked for is sent for you.",
826        json!({
827            "type": "send_stomp_message",
828            "destination": "/queue/test",
829            "subscription": "sub-0",
830            "message_id": "msg-1-retry",
831            "body": "redelivered after NACK",
832            "encoding": "utf8"
833        }),
834    )
835    .with_parameters(vec![Parameter {
836        name: "id".to_string(),
837        type_hint: "string".to_string(),
838        description: "The ack id, which is the message-id of the MESSAGE being rejected"
839            .to_string(),
840        required: true,
841    }])
842    .with_log_template(
843        LogTemplate::new()
844            .with_info("STOMP NACK id={id}")
845            .with_debug("STOMP nack: id={id}")
846            .with_trace("STOMP nack: {json_pretty(.)}"),
847    )
848    .with_actions(session_actions())
849});
850
851/// Client is closing the session.
852pub static STOMP_DISCONNECT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
853    EventType::new(
854        "stomp_disconnect",
855        "Client is shutting the session down. The connection closes as soon as this is handled \
856         (and after its receipt, if it asked for one), so this is the last chance to say \
857         anything.",
858        json!({"type": "close_connection"}),
859    )
860    .with_parameters(vec![Parameter {
861        name: "receipt".to_string(),
862        type_hint: "string".to_string(),
863        description: "The receipt id the client asked for, if any. It is answered \
864                      automatically; this field is here so a handler can see it."
865            .to_string(),
866        required: false,
867    }])
868    .with_log_template(
869        LogTemplate::new()
870            .with_info("STOMP DISCONNECT")
871            .with_debug("STOMP disconnect: receipt={receipt}")
872            .with_trace("STOMP disconnect: {json_pretty(.)}"),
873    )
874    .with_actions(vec![
875        send_stomp_message_action(),
876        send_stomp_receipt_action(),
877        close_connection_action(),
878    ])
879});
880
881pub fn get_stomp_event_types() -> Vec<EventType> {
882    vec![
883        STOMP_CONNECT_EVENT.clone(),
884        STOMP_SEND_EVENT.clone(),
885        STOMP_SUBSCRIBE_EVENT.clone(),
886        STOMP_UNSUBSCRIBE_EVENT.clone(),
887        STOMP_ACK_EVENT.clone(),
888        STOMP_NACK_EVENT.clone(),
889        STOMP_DISCONNECT_EVENT.clone(),
890    ]
891}
netget/server/stomp/actions.rs

netget/server/stomp/
actions.rs
