1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
// Example demonstrating GraphQL subscription support with call_tool_stream
use futures::{SinkExt, StreamExt};
use rs_utcp::UtcpClientInterface;
use serde_json::json;
use std::time::Duration;
use tokio::net::TcpListener;
use tokio::time::sleep;
use tokio_tungstenite::{accept_async, tungstenite::Message};
#[path = "common/mod.rs"]
mod common;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
// Spawn a GraphQL subscription server
let addr = spawn_graphql_subscription_server().await?;
println!("Started GraphQL subscription server at ws://{}", addr);
// Give the server time to start
sleep(Duration::from_millis(100)).await;
// Create a UTCP client with a GraphQL subscription provider
let client = common::client_from_providers(json!({
"manual_version": "1.0.0",
"utcp_version": "0.3.0",
"allowed_communication_protocols": ["graphql"],
"info": {
"title": "GraphQL Subscription Demo",
"version": "1.0.0",
"description": "GraphQL Subscription Demo Manual"
},
"tools": [{
"name": "stockPriceUpdates",
"description": "Stream stock price updates",
"inputs": { "type": "object" },
"outputs": { "type": "object" },
"tool_call_template": {
"call_template_type": "graphql",
"name": "stock_sub",
"url": format!("http://{}", addr),
"operation_type": "subscription"
}
}]
}))
.await?;
println!("Subscribing to stock price updates...");
// Call the subscription tool and get a stream
let mut stream = client
.call_tool_stream("stock_sub.stockPriceUpdates", Default::default())
.await?;
// Consume streaming results
let mut count = 0;
while let Ok(Some(value)) = stream.next().await {
println!("📈 Update #{}: {}", count + 1, value);
count += 1;
if count >= 5 {
// Stop after 5 updates
break;
}
}
println!("\n✅ Received {} stock price updates", count);
stream.close().await?;
Ok(())
}
/// Spawns a GraphQL subscription server that implements the graphql-transport-ws protocol
async fn spawn_graphql_subscription_server() -> anyhow::Result<std::net::SocketAddr> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
tokio::spawn(async move {
loop {
if let Ok((stream, _)) = listener.accept().await {
tokio::spawn(async move {
if let Ok(mut ws) = accept_async(stream).await {
// Handle GraphQL subscription protocol
while let Some(msg) = ws.next().await {
match msg {
Ok(Message::Text(text)) => {
let payload: serde_json::Value =
serde_json::from_str(&text).unwrap_or_default();
match payload.get("type").and_then(|v| v.as_str()) {
Some("connection_init") => {
// Send connection_ack
let _ = ws
.send(Message::Text(
json!({ "type": "connection_ack" }).to_string(),
))
.await;
}
Some("subscribe") => {
// Send periodic stock price updates
for i in 1..=10 {
let price = 100.0 + (i as f64 * 2.5);
let update = json!({
"id": "1",
"type": "next",
"payload": {
"data": {
"stockPriceUpdates": {
"symbol": "UTCP",
"price": price,
"update": i
}
}
}
});
if ws
.send(Message::Text(update.to_string()))
.await
.is_err()
{
return;
}
sleep(Duration::from_secs(1)).await;
}
// Send complete message
let _ = ws
.send(Message::Text(
json!({
"id": "1",
"type": "complete"
})
.to_string(),
))
.await;
return;
}
_ => {}
}
}
Ok(Message::Close(_)) => return,
Err(_) => return,
_ => {}
}
}
}
});
}
}
});
Ok(addr)
}