1use crate::llm::actions::{
4 client_trait::{Client, ClientActionResult},
5 protocol_trait::Protocol,
6 ActionDefinition, ParameterDefinition, StartupExamples,
7};
8use crate::protocol::{ConnectContext, EventType};
9use crate::server::pulsar::actions::{action, check_message, message_params, p};
10use crate::server::pulsar::wire;
11use crate::state::app_state::AppState;
12use anyhow::{bail, ensure, Context, Result};
13use serde_json::{json, Value};
14use std::sync::LazyLock;
15
16pub const PRODUCE: &str = "pulsar_produce";
17pub const SUBSCRIBE: &str = "pulsar_subscribe";
18pub const UNSUBSCRIBE: &str = "pulsar_unsubscribe";
19
20#[derive(Default)]
21pub struct PulsarClientProtocol;
22impl PulsarClientProtocol {
23 pub fn new() -> Self {
24 Self
25 }
26}
27
28fn topic_sub() -> Vec<crate::llm::actions::Parameter> {
29 vec![
30 p(
31 "topic",
32 "string",
33 "Topic name, e.g. orders or persistent://public/default/orders",
34 true,
35 ),
36 p(
37 "subscription",
38 "string",
39 "Subscription name; consumers sharing it share its messages",
40 true,
41 ),
42 ]
43}
44
45pub fn actions() -> Vec<ActionDefinition> {
46 let mut sub = topic_sub();
47 sub.push(p(
48 "sub_type",
49 "string",
50 "Exclusive (default), Shared, Failover or Key_Shared",
51 false,
52 ));
53 sub.push(p(
54 "initial_position",
55 "string",
56 "Where a new subscription starts: latest (default) or earliest",
57 false,
58 ));
59 vec![
60 action(PRODUCE, "Publish a message to a topic; the broker's receipt (or refusal) arrives as pulsar_produced.",
61 message_params(), json!({"type": PRODUCE, "topic": "orders", "payload": "order 42 shipped", "properties": {"source": "netget"}})),
62 action(SUBSCRIBE, "Subscribe to a topic; every message that arrives is a pulsar_message event and is acknowledged once handled.",
63 sub, json!({"type": SUBSCRIBE, "topic": "orders", "subscription": "netget"})),
64 action(UNSUBSCRIBE, "Delete a subscription this client holds.",
65 topic_sub(), json!({"type": UNSUBSCRIBE, "topic": "orders", "subscription": "netget"})),
66 action("disconnect", "Close the connection to the broker.", vec![], json!({"type": "disconnect"})),
67 ]
68}
69
70fn event(id: &str, description: &str, params: Vec<crate::llm::actions::Parameter>) -> EventType {
71 EventType::new(
72 id,
73 description,
74 json!({"type": PRODUCE, "topic": "orders", "payload": "hello"}),
75 )
76 .with_parameters(params)
77 .with_actions(actions())
78}
79
80pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
81 event(
82 "pulsar_connected",
83 "Connected to the broker (CONNECT answered).",
84 vec![
85 p(
86 "server_version",
87 "string",
88 "What the broker said it is",
89 true,
90 ),
91 p(
92 "protocol_version",
93 "number",
94 "The protocol version agreed",
95 true,
96 ),
97 p(
98 "max_message_size",
99 "number",
100 "The largest message the broker takes, in bytes",
101 false,
102 ),
103 ],
104 )
105});
106
107pub static PRODUCED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
108 event(
109 "pulsar_produced",
110 "The broker answered a published message.",
111 vec![
112 p("topic", "string", "The full topic name", true),
113 p("ok", "boolean", "Whether the broker stored it", true),
114 p(
115 "message_id",
116 "object",
117 "Its id {ledger_id, entry_id} when stored",
118 false,
119 ),
120 p(
121 "error",
122 "string",
123 "Why the broker refused it (or the producer)",
124 false,
125 ),
126 ],
127 )
128});
129
130pub static SUBSCRIBED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
131 event(
132 "pulsar_subscribed",
133 "The broker answered a subscribe or unsubscribe.",
134 vec![
135 p("topic", "string", "The full topic name", true),
136 p("subscription", "string", "The subscription name", true),
137 p("operation", "string", "subscribe or unsubscribe", true),
138 p("ok", "boolean", "Whether it succeeded", true),
139 p("error", "string", "Why the broker refused", false),
140 ],
141 )
142});
143
144pub static MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
145 event(
146 "pulsar_message",
147 "A message arrived on a subscription (it is acknowledged once this event is handled).",
148 vec![
149 p("topic", "string", "The full topic name", true),
150 p(
151 "subscription",
152 "string",
153 "The subscription it arrived on",
154 true,
155 ),
156 p(
157 "payload",
158 "string",
159 "The message body as text, or hex when encoding is hex",
160 true,
161 ),
162 p(
163 "encoding",
164 "string",
165 "How payload is written: utf8, or hex when the body is not printable text",
166 true,
167 ),
168 p(
169 "properties",
170 "object",
171 "The message's string properties",
172 true,
173 ),
174 p("key", "string", "The message key, if any", false),
175 p(
176 "producer_name",
177 "string",
178 "Which producer published it",
179 true,
180 ),
181 p("message_id", "object", "Its id {ledger_id, entry_id}", true),
182 p(
183 "redelivery_count",
184 "number",
185 "How many times it was delivered before",
186 true,
187 ),
188 ],
189 )
190});
191
192pub fn check(v: &Value) -> Result<()> {
193 match v["type"].as_str().unwrap_or_default() {
194 PRODUCE => check_message(v),
195 SUBSCRIBE | UNSUBSCRIBE => {
196 wire::full_topic(v["topic"].as_str().context("topic is required")?)?;
197 let s = v["subscription"]
198 .as_str()
199 .context("subscription is required")?;
200 ensure!(
201 !s.is_empty() && s.len() <= 256,
202 "subscription must be 1-256 bytes"
203 );
204 if let Some(t) = v["sub_type"].as_str() {
205 ensure!(
206 wire::SUB_TYPES.contains(&t),
207 "sub_type must be one of {:?}",
208 wire::SUB_TYPES
209 );
210 }
211 if let Some(p) = v["initial_position"].as_str() {
212 ensure!(
213 matches!(p, "latest" | "earliest"),
214 "initial_position must be latest or earliest"
215 );
216 }
217 Ok(())
218 }
219 other => bail!("Unknown Pulsar client action {other:?}"),
220 }
221}
222
223impl Protocol for PulsarClientProtocol {
224 fn protocol_name(&self) -> &'static str {
225 "Apache Pulsar"
226 }
227 fn stack_name(&self) -> &'static str {
228 "ETH>IP>TCP>Pulsar"
229 }
230 fn keywords(&self) -> Vec<&'static str> {
231 vec![
232 "pulsar",
233 "apache pulsar",
234 "pulsar client",
235 "pulsar producer",
236 "pulsar consumer",
237 ]
238 }
239 fn description(&self) -> &'static str {
240 "Apache Pulsar client (binary protocol): publishes, subscribes and hears messages as the model directs"
241 }
242 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
243 actions()
244 }
245 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
246 vec![]
247 }
248 fn get_event_types(&self) -> Vec<EventType> {
249 vec![
250 CONNECTED_EVENT.clone(),
251 PRODUCED_EVENT.clone(),
252 SUBSCRIBED_EVENT.clone(),
253 MESSAGE_EVENT.clone(),
254 ]
255 }
256 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
257 Vec::new()
258 }
259 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
260 use crate::protocol::metadata::*;
261 ProtocolMetadataV2::builder()
262 .state(DevelopmentState::Experimental)
263 .privilege_requirement(PrivilegeRequirement::None)
264 .implementation("The Pulsar binary protocol over Tokio TCP, sharing the broker role's prost messages and framing: CONNECT, LOOKUP before each producer and subscription, PRODUCER/SEND with CRC32C, SUBSCRIBE/FLOW with permits topped up, MESSAGE (batches split) acknowledged after the handler runs, PING answered")
265 .llm_control("What to publish where, which topics to subscribe to, and what to do with each message")
266 .e2e_testing("tests/client/pulsar: Apache Pulsar 4.0.6 standalone; the official Python client reads what NetGet publishes and publishes what NetGet receives")
267 .notes("A lookup that redirects to another broker is reported, not followed. No partitioned topics, schemas, compression, encryption, transactions or TLS; acknowledgement is automatic. A handler chain stops after 8 follow-ups.")
268 .build()
269 }
270 fn example_prompt(&self) -> &'static str {
271 "Connect to the Pulsar broker at 127.0.0.1:6650, subscribe to orders and echo each one to receipts"
272 }
273 fn get_startup_examples(&self) -> StartupExamples {
274 let llm = json!({"type":"open_client","protocol":"pulsar","remote_addr":"127.0.0.1:6650",
275 "instruction":"Subscribe to orders and publish a receipt to receipts for each order"});
276 let mut static_example = llm.clone();
277 static_example["event_handlers"] = json!([{"event_pattern":"pulsar_connected","handler":{"type":"static","actions":[{"type":SUBSCRIBE,"topic":"orders","subscription":"netget"}]}}]);
278 let mut scripted = llm.clone();
279 scripted["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python",
280 "code":"import json,sys\ni=json.load(sys.stdin); t=i['event_type_id']; e=i['event']\na=[]\nif t=='pulsar_connected': a=[{'type':'pulsar_subscribe','topic':'orders','subscription':'netget'}]\nelif t=='pulsar_message': a=[{'type':'pulsar_produce','topic':'receipts','payload':'got '+e['payload']}]\nprint(json.dumps({'actions':a}))"}}]);
281 StartupExamples::new(llm, scripted, static_example)
282 }
283 fn group_name(&self) -> &'static str {
284 "Application"
285 }
286}
287
288impl Client for PulsarClientProtocol {
289 fn connect(
290 &self,
291 ctx: ConnectContext,
292 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
293 {
294 Box::pin(super::connect(ctx))
295 }
296 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
297 let name = v["type"].as_str().unwrap_or_default().to_string();
298 if name == "disconnect" {
299 return Ok(ClientActionResult::Disconnect);
300 }
301 check(&v)?;
302 Ok(ClientActionResult::Custom { name, data: v })
303 }
304}