netget/server/pulsar/
actions.rs

1//! What the model decides as a Pulsar broker: which producers and subscriptions it allows,
2//! and whether each published message is accepted (and delivered to the topic's
3//! subscriptions) or refused. Rust owns the sessions, lookups, flow control, delivery and
4//! acknowledgement.
5use crate::llm::actions::{
6    protocol_trait::{ActionResult, Protocol, Server},
7    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
8};
9use crate::protocol::log_template::LogTemplate;
10use crate::protocol::{EventType, SpawnContext};
11use crate::state::app_state::AppState;
12use anyhow::{bail, ensure, Context, Result};
13use serde_json::{json, Value};
14use std::sync::LazyLock;
15
16pub const ACCEPT: &str = "pulsar_accept";
17pub const REJECT: &str = "pulsar_reject";
18pub const PUBLISH: &str = "pulsar_publish";
19/// Properties one message may carry.
20pub const MAX_PROPERTIES: usize = 64;
21
22#[derive(Default)]
23pub struct PulsarProtocol;
24impl PulsarProtocol {
25    pub fn new() -> Self {
26        Self
27    }
28}
29
30pub fn p(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
31    Parameter {
32        name: name.into(),
33        type_hint: kind.into(),
34        description: description.into(),
35        required,
36    }
37}
38
39pub fn action(
40    name: &str,
41    description: &str,
42    parameters: Vec<Parameter>,
43    example: Value,
44) -> ActionDefinition {
45    ActionDefinition {
46        name: name.into(),
47        description: description.into(),
48        parameters,
49        example,
50        log_template: Some(LogTemplate::new().with_info(format!("-> Pulsar {name}"))),
51    }
52}
53
54fn accept() -> ActionDefinition {
55    action(ACCEPT, "Allow the producer or subscription, or accept the message (it is stored for and delivered to the topic's subscriptions).",
56        vec![], json!({"type": ACCEPT}))
57}
58
59fn reject() -> ActionDefinition {
60    action(REJECT, "Refuse the producer, subscription or message; the client gets an authorization or not-allowed error with this text.",
61        vec![p("message", "string", "Why, as the client will see it", true)],
62        json!({"type": REJECT, "message": "not allowed on this topic"}))
63}
64
65/// The parameters a message is described with, in both roles.
66pub fn message_params() -> Vec<Parameter> {
67    vec![
68        p(
69            "topic",
70            "string",
71            "Topic name, e.g. orders or persistent://public/default/orders",
72            true,
73        ),
74        p(
75            "payload",
76            "string",
77            "The message body as text, or hex when encoding is hex",
78            true,
79        ),
80        p(
81            "encoding",
82            "string",
83            "How payload is written: utf8 (default) or hex",
84            false,
85        ),
86        p(
87            "properties",
88            "object",
89            "String properties, e.g. {\"source\": \"netget\"}",
90            false,
91        ),
92        p("key", "string", "The message key (partition key)", false),
93    ]
94}
95
96pub fn publish() -> ActionDefinition {
97    action(PUBLISH, "Publish a message to a topic as producer netget, delivered to its subscriptions like any other.",
98        message_params(), json!({"type": PUBLISH, "topic": "replies", "payload": "got it", "properties": {"source": "netget"}}))
99}
100
101pub fn all_actions() -> Vec<ActionDefinition> {
102    vec![accept(), reject(), publish()]
103}
104
105fn event(id: &str, description: &str, params: Vec<Parameter>) -> EventType {
106    EventType::new(id, description, json!({"type": ACCEPT}))
107        .with_parameters(params)
108        .with_actions(all_actions())
109}
110
111pub static PRODUCER_EVENT: LazyLock<EventType> = LazyLock::new(|| {
112    event(
113        "pulsar_producer",
114        "A client asked to publish to a topic. Accept or reject the producer.",
115        vec![
116            p("topic", "string", "The full topic name", true),
117            p(
118                "producer_name",
119                "string",
120                "The producer's name (NetGet assigns one when the client gives none)",
121                true,
122            ),
123            p(
124                "access_mode",
125                "string",
126                "Shared, Exclusive, WaitForExclusive or ExclusiveWithFencing",
127                true,
128            ),
129            p(
130                "remote_addr",
131                "string",
132                "The client's address and port",
133                true,
134            ),
135        ],
136    )
137});
138
139pub static SUBSCRIBE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
140    event(
141        "pulsar_subscribe",
142        "A client asked to consume from a topic. Accept or reject the subscription.",
143        vec![
144            p("topic", "string", "The full topic name", true),
145            p("subscription", "string", "The subscription name", true),
146            p(
147                "sub_type",
148                "string",
149                "Exclusive, Shared, Failover or Key_Shared",
150                true,
151            ),
152            p(
153                "consumer_name",
154                "string",
155                "The consumer's name, if it gave one",
156                false,
157            ),
158            p(
159                "remote_addr",
160                "string",
161                "The client's address and port",
162                true,
163            ),
164        ],
165    )
166});
167
168pub static MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
169    let mut params = vec![
170        p("producer_name", "string", "Which producer sent it", true),
171        p(
172            "sequence_id",
173            "number",
174            "The producer's sequence number for it",
175            true,
176        ),
177        p(
178            "event_time",
179            "number",
180            "Event time in Unix milliseconds, if the producer set one",
181            false,
182        ),
183        p(
184            "remote_addr",
185            "string",
186            "The client's address and port",
187            true,
188        ),
189    ];
190    params.extend(message_params());
191    event("pulsar_message", "A producer published a message. Accept it (it is delivered to the topic's subscriptions) or reject it.", params)
192});
193
194pub fn check_message(v: &Value) -> Result<()> {
195    super::wire::full_topic(v["topic"].as_str().context("topic is required")?)?;
196    let payload = v["payload"].as_str().context("payload is required")?;
197    let bytes = super::wire::bytes(payload, v["encoding"].as_str())?;
198    ensure!(
199        bytes.len() <= super::wire::MAX_FRAME / 2,
200        "payload too large"
201    );
202    if let Some(props) = v.get("properties").filter(|x| !x.is_null()) {
203        let o = props
204            .as_object()
205            .context("properties must be an object of strings")?;
206        ensure!(
207            o.len() <= MAX_PROPERTIES,
208            "at most {MAX_PROPERTIES} properties"
209        );
210        ensure!(
211            o.values().all(Value::is_string),
212            "property values must be strings"
213        );
214    }
215    if let Some(k) = v.get("key").filter(|x| !x.is_null()) {
216        ensure!(k.is_string(), "key must be a string");
217    }
218    Ok(())
219}
220
221pub fn check(v: &Value) -> Result<()> {
222    match v["type"].as_str().unwrap_or_default() {
223        ACCEPT => Ok(()),
224        REJECT => {
225            let m = v["message"].as_str().context("message is required")?;
226            ensure!(
227                !m.is_empty() && m.len() <= 512,
228                "message must be 1-512 bytes"
229            );
230            Ok(())
231        }
232        PUBLISH => check_message(v),
233        other => bail!("Unknown Pulsar action {other:?}"),
234    }
235}
236
237impl Protocol for PulsarProtocol {
238    fn protocol_name(&self) -> &'static str {
239        "Apache Pulsar"
240    }
241    fn stack_name(&self) -> &'static str {
242        "ETH>IP>TCP>Pulsar"
243    }
244    fn keywords(&self) -> Vec<&'static str> {
245        vec![
246            "pulsar",
247            "apache pulsar",
248            "pulsar broker",
249            "pub/sub",
250            "message broker",
251        ]
252    }
253    fn description(&self) -> &'static str {
254        "Apache Pulsar broker (binary protocol): the model allows producers and subscriptions and accepts or refuses each message; Rust delivers to subscriptions"
255    }
256    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
257        Vec::new()
258    }
259    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
260        all_actions()
261    }
262    fn get_event_types(&self) -> Vec<EventType> {
263        vec![
264            PRODUCER_EVENT.clone(),
265            SUBSCRIBE_EVENT.clone(),
266            MESSAGE_EVENT.clone(),
267        ]
268    }
269    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
270        vec![ParameterDefinition {
271            name: "idle_timeout_secs".into(),
272            type_hint: "number".into(),
273            description: "Seconds a client may send nothing (clients ping every 30) before it is disconnected (1..=86400)".into(),
274            required: false,
275            example: json!(600),
276            default: Some(json!(super::IDLE_TIMEOUT.as_secs())),
277        }]
278    }
279    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
280        use crate::protocol::metadata::*;
281        ProtocolMetadataV2::builder()
282            .state(DevelopmentState::Experimental)
283            .privilege_requirement(PrivilegeRequirement::None)
284            .well_known_port(6650)
285            .implementation("The Pulsar binary protocol over Tokio TCP with hand-declared prost messages: CONNECT, PARTITIONED_METADATA (non-partitioned), LOOKUP (to itself), PRODUCER, SEND (CRC32C-checked, batches split), SUBSCRIBE (Exclusive, Shared, Failover, Key_Shared as Shared), FLOW, MESSAGE, ACK (individual, cumulative, with receipts), REDELIVER_UNACKNOWLEDGED_MESSAGES, UNSUBSCRIBE, CLOSE_*, GET_LAST_MESSAGE_ID, PING/PONG")
286            .llm_control("Which producers and subscriptions are allowed, whether each message is accepted, and messages it publishes itself")
287            .e2e_testing("tests/server/pulsar: the official Python client (pulsar-client, the C++ library) and the Java CLI from Pulsar 4.0.6 produce and consume through NetGet; raw frames for checksums, bounds and a failed handler")
288            .notes("Messages live in memory while the server runs: a subscription keeps a bounded backlog (10000) for consumers without permits, and a new subscription starts at the latest message whatever position it asks for. No persistence, partitions, schemas, compression, encryption, chunking, transactions, authentication or TLS. Frames are capped at 5 MiB.")
289            .request_only("Every command answers a client's request, except MESSAGE frames delivered to consumers")
290            .answers_on_failure()
291            .max_inbound_bytes(super::wire::MAX_FRAME)
292            .build()
293    }
294    fn example_prompt(&self) -> &'static str {
295        "Pulsar broker on port 6650 that refuses messages containing passwords"
296    }
297    fn get_startup_examples(&self) -> StartupExamples {
298        let llm = json!({"type":"open_server","base_stack":"pulsar","port":6650,
299            "instruction":"Allow every producer and subscription; refuse messages whose payload mentions a password"});
300        let mut static_example = llm.clone();
301        static_example["event_handlers"] =
302            json!([{"event_pattern":"*","handler":{"type":"static","actions":[{"type":ACCEPT}]}}]);
303        let mut scripted = llm.clone();
304        scripted["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python",
305            "code":"import json,sys\ni=json.load(sys.stdin); e=i['event']\nbad=i['event_type_id']=='pulsar_message' and 'password' in e['payload']\nprint(json.dumps({'actions':[{'type':'pulsar_reject','message':'no secrets'} if bad else {'type':'pulsar_accept'}]}))"}}]);
306        StartupExamples::new(llm, scripted, static_example)
307    }
308    fn group_name(&self) -> &'static str {
309        "Application"
310    }
311}
312
313impl Server for PulsarProtocol {
314    fn spawn(
315        &self,
316        ctx: SpawnContext,
317    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
318    {
319        Box::pin(super::spawn(ctx))
320    }
321    fn execute_action(&self, action: Value) -> Result<ActionResult> {
322        check(&action)?;
323        Ok(ActionResult::Custom {
324            name: action["type"].as_str().unwrap_or_default().to_string(),
325            data: action,
326        })
327    }
328}
netget/server/pulsar/actions.rs

netget/server/pulsar/
actions.rs
