rpi-server

TCP JSONL server for rpi with streaming subscriptions — 基于 Transport trait 抽象，支持多种传输协议。

特性

✅ TCP JSONL 协议 — 简单、高效、易于调试
✅ Transport trait 抽象 — 可扩展 WebSocket/Unix Socket 等传输方式
✅ 流式订阅 — 通过 subscribe 方法订阅实时事件流
✅ 会话管理 — 支持多个 rpi 会话并行
✅ 连接管理 — 内置重试、心跳、延迟检测、在线状态追踪
✅ 标准化 — JSON-RPC 2.0 消息格式，跨语言兼容

架构

┌─────────────────────────────────────────┐
│         Transport trait                 │
│  ┌─────────────────────────────────┐   │
│  │   TcpTransport (已实现)          │   │
│  │   - TCP JSONL 协议               │   │
│  │   - 服务间通信首选               │   │
│  └─────────────────────────────────┘   │
│  ┌─────────────────────────────────┐   │
│  │   WebSocketTransport (未来)      │   │
│  │   - 浏览器/Web 客户端            │   │
│  │   - HTTP 升级协议                │   │
│  └─────────────────────────────────┘   │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│      ManagedConnection                  │
│  - 自动重试（最多 10 次）               │
│  - 心跳检测（30s 间隔）                 │
│  - 延迟测量                             │
│  - 在线状态追踪                         │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│         AppState + Sessions             │
│  - 会话生命周期管理                     │
│  - 子进程管理                           │
│  - 事件广播                             │
└─────────────────────────────────────────┘

安装

cargo build -p rpi-server --release

使用方式

独立运行

# 启动服务器（默认 127.0.0.1:9800）
rpi-server

# 无 TUI 模式（明确指定）
rpi-server --server

# 指定端口
rpi-server --port 8080

# 传递 rpi 参数
rpi-server -- --provider anthropic --model claude-3-sonnet

作为 rpi 插件

// 启动服务器
{ "action": "start", "port": 9800 }

// 响应
{
  "id": "rpi-rpc-1",
  "address": "127.0.0.1:9800",
  "token": "a1b2c3d4...",
  "state": "running",
  "protocol": "jsonl",
  "transport": "tcp"
}

客户端连接

使用任何 TCP 客户端连接到服务器，发送 JSON-RPC 2.0 格式的请求：

# 使用 nc 测试
nc 127.0.0.1 9800

# 发送启动会话请求
{"jsonrpc":"2.0","id":1,"method":"start_session","params":{}}

# 响应
{"jsonrpc":"2.0","id":1,"result":{"sessionId":"session-1","session":{...}}}

使用 ManagedConnection（推荐）

use std::sync::Arc;
use rpi_server::transport::{ManagedConnection, TcpTransport};

let transport = Arc::new(TcpTransport::new());
let conn = ManagedConnection::new(transport);

// 自动重试连接（最多 10 次）
conn.connect_with_retry("127.0.0.1:9800").await?;

// 发送请求
conn.send(json!({
    "jsonrpc": "2.0",
    "id": 1,
    "method": "start_session",
    "params": {}
})).await?;

// 接收响应
if let Some(response) = conn.recv().await? {
    println!("response: {}", response);
}

// 获取连接状态
let status = conn.status().await;
println!("connected: {}, latency: {}ms", status.connected, status.latency_ms);

// 启动心跳（在独立 task 中）
let conn_clone = conn.clone();
tokio::spawn(async move {
    let _ = conn_clone.start_heartbeat().await;
});

JSON-RPC 方法

`start_session`

启动新的 rpi 会话。

{
  "jsonrpc": "2.0",
  "method": "start_session",
  "params": {},
  "id": 1
}

响应：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "sessionId": "session-1",
    "session": {
      "id": "session-1",
      "state": "running",
      "pid": 12345,
      "event_count": 0
    }
  }
}

`send`

向会话发送命令。

{
  "jsonrpc": "2.0",
  "method": "send",
  "params": {
    "sessionId": "session-1",
    "command": {"type": "prompt", "content": "hello"}
  },
  "id": 2
}

`subscribe`

订阅会话的事件流（流式响应）。

{
  "jsonrpc": "2.0",
  "method": "subscribe",
  "params": {"sessionId": "session-1"},
  "id": 3
}

响应：

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "subscriptionId": 1,
    "sessionId": "session-1",
    "status": "subscribed"
  }
}

后续事件流：

{"jsonrpc":"2.0","method":"event","params":{"subscriptionId":1,"event":{"type":"delta","content":"Hello"}}}
{"jsonrpc":"2.0","method":"event","params":{"subscriptionId":1,"event":{"type":"delta","content":" world"}}}
{"jsonrpc":"2.0","method":"event","params":{"subscriptionId":1,"event":{"type":"session_end"}}}

`stop_session`

停止会话。

{
  "jsonrpc": "2.0",
  "method": "stop_session",
  "params": {"sessionId": "session-1"},
  "id": 4
}

`list_sessions`

列出所有会话。

{
  "jsonrpc": "2.0",
  "method": "list_sessions",
  "params": {},
  "id": 5
}

`server_status`

获取服务器状态。

{
  "jsonrpc": "2.0",
  "method": "server_status",
  "params": {},
  "id": 6
}

`ping`

心跳检测。

{
  "jsonrpc": "2.0",
  "method": "ping",
  "params": {},
  "id": 7
}

响应：

{
  "jsonrpc": "2.0",
  "id": 7,
  "result": {"type": "pong"}
}

跨语言客户端

由于使用标准 JSON-RPC 2.0 over TCP，可以用任何语言调用：

Python

import socket
import json

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 9800))

# 发送请求
request = {
    "jsonrpc": "2.0",
    "method": "start_session",
    "params": {},
    "id": 1
}
sock.sendall((json.dumps(request) + '\n').encode())

# 接收响应
buffer = b''
while True:
    chunk = sock.recv(4096)
    if not chunk:
        break
    buffer += chunk
    if b'\n' in buffer:
        line, buffer = buffer.split(b'\n', 1)
        response = json.loads(line)
        print(response)
        break

Node.js

const net = require('net');

const client = new net.Socket();
client.connect(9800, '127.0.0.1', () => {
    const request = {
        jsonrpc: '2.0',
        method: 'start_session',
        params: {},
        id: 1
    };
    client.write(JSON.stringify(request) + '\n');
});

client.on('data', (data) => {
    const lines = data.toString().split('\n').filter(l => l.trim());
    for (const line of lines) {
        const response = JSON.parse(line);
        console.log(response);
    }
});

Transport trait 扩展

要添加新的传输协议（如 WebSocket），只需实现 Transport 和 Connection trait：

use async_trait::async_trait;
use rpi_server::transport::{Transport, Connection};

pub struct WebSocketTransport {
    // ...
}

#[async_trait]
impl Transport for WebSocketTransport {
    type Connection = WebSocketConnection;
    
    async fn listen(&self, addr: &str) -> Result<(), String> {
        // 实现 WebSocket 监听
    }
    
    async fn accept(&self) -> Result<WebSocketConnection, String> {
        // 实现接受连接
    }
    
    async fn connect(&self, addr: &str) -> Result<WebSocketConnection, String> {
        // 实现客户端连接
    }
    
    async fn shutdown(&self) -> Result<(), String> {
        // 实现关闭
    }
}

pub struct WebSocketConnection {
    // ...
}

#[async_trait]
impl Connection for WebSocketConnection {
    async fn send(&mut self, msg: serde_json::Value) -> Result<(), String> {
        // 实现发送
    }
    
    async fn recv(&mut self) -> Result<Option<serde_json::Value>, String> {
        // 实现接收
    }
    
    fn id(&self) -> &str {
        // 返回连接 ID
    }
    
    async fn close(&mut self) -> Result<(), String> {
        // 实现关闭
    }
}

然后在 run_server 中使用：

let transport = Arc::new(WebSocketTransport::new());
run_server(transport, "127.0.0.1", 9800, token, executable, args).await?;

连接管理

ManagedConnection 提供以下功能：

自动重试

// 连接失败时自动重试，最多 10 次，间隔递增
conn.connect_with_retry("127.0.0.1:9800").await?;

心跳检测

// 启动心跳任务（30s 间隔）
let conn_clone = conn.clone();
tokio::spawn(async move {
    let _ = conn_clone.start_heartbeat().await;
});

状态追踪

let status = conn.status().await;
println!("connected: {}", status.connected);
println!("latency: {}ms", status.latency_ms);
println!("messages sent: {}", status.messages_sent);
println!("messages received: {}", status.messages_received);

性能

TCP JSONL 比 WebSocket 少一层协议开销，但需要自己处理消息边界（\n）
broadcast channel 支持多个订阅者同时接收事件
异步 I/O — 基于 tokio，高并发性能优秀
零拷贝 — 消息直接序列化到 TCP 流，无中间缓冲

与 WebSocket 对比

特性	TCP JSONL	WebSocket
协议开销	低（仅 `\n` 分隔）	中（帧协议）
浏览器支持	❌ 需要代理	✅ 原生支持
调试难度	低（nc/telnet）	中（需要工具）
心跳检测	应用层实现	协议自带
适用场景	服务间通信	Web 客户端

License

MIT

rpi-server 0.2.0

rpi-server

特性

架构

安装

使用方式

独立运行

作为 rpi 插件

客户端连接

使用 ManagedConnection（推荐）

JSON-RPC 方法

start_session

send

subscribe

stop_session

list_sessions

server_status

ping

跨语言客户端

Python

Node.js

Transport trait 扩展

连接管理

自动重试

心跳检测

状态追踪

性能

与 WebSocket 对比

License

`start_session`

`send`

`subscribe`

`stop_session`

`list_sessions`

`server_status`

`ping`