Compare commits

..
2 Commits
Author SHA1 Message Date
qing bf2b5d31f7 fix(botops): v1.2 - 彻底修复思考区域与正式回答顺序问题
问题根因:之前预创建空 AI widget,思考 widget 挂载在其后,即使动态重排也可能因 Textual 组件树更新时序导致顺序不稳定。

修复方案:
- 移除 prefab 空 AI widget 预创建逻辑
- start_ai() 仅标记状态,不创建任何 widget
- _ensure_ai_widget() 延迟到第一次 chunk 到来时才创建 AI widget
- 按消息到达顺序自然挂载:reasoning_start → reasoning → chunk → end_reasoning/finish_ai
- 思考 widget 必然先于 AI widget 挂载,顺序天然正确

同时:
- 简化 app.py 发送失败处理,不再需要移除空 widget
- 更新 README v1.2 说明
2026-08-01 05:00:56 +08:00
qing 9815ff44af feat(libhttp): integrate WebSocket client into HttpClient
Merge WebSocket client functionality into HttpClient using Boost.Beast,
enabling a single client instance to handle both HTTP (libcurl) and
WebSocket (ws:// and wss://) protocols. Backward-compatible type aliases
(WebSocketClient, WebSocketClientBase) preserve existing code.

Changes:
- Add synchronous WebSocket client: ws_connect/ws_send/ws_recv/ws_close
- Support ws:// (plain TCP) and wss:// (TLS with SNI) connections
- Layered stream architecture with beast::tcp_stream for timeout support
- Binary message support via ws_send_binary()
- Update Makefile to link libcurl (-lcurl)
- Add README.md documenting unified HttpServer + HttpClient API
2026-08-01 04:59:47 +08:00
6 changed files with 542 additions and 74 deletions

No files matched your search

+6 -4
View File
@@ -63,20 +63,22 @@ BotOps/
- `reasoning_start`(**思考开始**):创建灰色斜体思考区域,实时展示 AI 思考过程 - `reasoning_start`(**思考开始**):创建灰色斜体思考区域,实时展示 AI 思考过程
- `reasoning`(**思考流**):思考内容增量片段,追加到思考区域(超长自动折叠显示最后 3000 字符) - `reasoning`(**思考流**):思考内容增量片段,追加到思考区域(超长自动折叠显示最后 3000 字符)
- `reasoning_end`(**思考结束**):思考结束,切换到正式回答 - `reasoning_end`(**思考结束**):思考结束,切换到正式回答
- **顺序保证**:通过动态重排 widget,思考区域始终显示在 AI 正式回答**之前** - **顺序保证**:不预创建空 AI widget,按消息到达顺序自然挂载(`reasoning_start`→`reasoning`→`chunk`),思考区域必然在 AI 正式回答**之前**
- `complete`(**收尾信号**):chunk 流式已完整渲染,complete 仅用于收尾;仅在完全丢帧(未收到任何 chunk)时用 full_text 补建消息,**不再做全文替换对齐**(避免服务端递归调用工具后 complete 只含最后一轮内容导致重复渲染) - `complete`(**收尾信号**):chunk 流式已完整渲染,complete 仅用于收尾(重置状态)
- `line`(过程行):🔧 工具调用提示、工具输出,路由到任务侧栏;助手前缀/思考提示在聊天区已由对应区域承载,不重复展示 - `line`(过程行):🔧 工具调用提示、工具输出,路由到任务侧栏;助手前缀/思考提示在聊天区已由对应区域承载,不重复展示
- `info`(系统信息):过滤"Conversation started"/"对话开始"等隐式通知,避免污染聊天流 - `info`(系统信息):过滤"Conversation started"/"对话开始"等隐式通知,避免污染聊天流
- `dialogue_ended` / `dialogue_reset` / `task_added` / `task_process` / `task_status` / `task_error` / `interrupt_ack` / `error` / `command_result` / `connection_ack` / `notification` / `pong` - `dialogue_ended` / `dialogue_reset` / `task_added` / `task_process` / `task_status` / `task_error` / `interrupt_ack` / `error` / `command_result` / `connection_ack` / `notification` / `pong`
## 更新日志 ## 更新日志
### v1.2
- ✅ **彻底修复思考区域顺序问题**:移除预创建空 AI widget 的逻辑,改为第一次收到 `chunk` 时才创建 widget。按消息到达顺序自然挂载(`reasoning_start`→`reasoning`→`chunk`),从根本上保证思考区域始终在正式回答之前
- ✅ 简化 `complete` 处理逻辑:仅作为收尾重置状态信号,不再进行补建逻辑
### v1.1 ### v1.1
- ✅ 支持 DeepSeek 思考模式展示:新增 `reasoning_start`/`reasoning`/`reasoning_end` 消息处理,思考内容以灰色斜体区域展示 - ✅ 支持 DeepSeek 思考模式展示:新增 `reasoning_start`/`reasoning`/`reasoning_end` 消息处理,思考内容以灰色斜体区域展示
- ✅ **修复思考区域顺序问题**:通过 widget 动态重排,保证思考过程始终显示在 AI 正式回答之前(而非之后)
- ✅ 超长思考内容自动折叠(显示最后 3000 字符) - ✅ 超长思考内容自动折叠(显示最后 3000 字符)
- ✅ 过滤"Conversation started"/"对话开始"等隐式系统通知,避免在回答末尾显示无用信息 - ✅ 过滤"Conversation started"/"对话开始"等隐式系统通知,避免在回答末尾显示无用信息
- ✅ 即使服务端未显式启用思考模式,若模型返回 reasoning_content 也能自动展示思考过程
### v1.0 ### v1.0
- 初始版本:Markdown 增量流式渲染、任务独立侧栏、自动重连、快捷键支持 - 初始版本:Markdown 增量流式渲染、任务独立侧栏、自动重连、快捷键支持
+2 -2
View File
@@ -147,9 +147,9 @@ class BotOpsApp(App):
await self.notify_system("未连接到服务器,先 /connect", level="error") await self.notify_system("未连接到服务器,先 /connect", level="error")
return return
await self.chat.add_user(text) await self.chat.add_user(text)
await self.chat.start_ai() # 预建 AI 消息框,即时反馈 await self.chat.start_ai() # 标记AI回复开始状态(不预创建空widget)
if not await self.ws.send_message(text): if not await self.ws.send_message(text):
await self.chat.finish_ai("") # 发送失败:移除预建的空消息框 await self.chat.finish_ai("") # 发送失败:重置AI状态
await self.notify_system("❌ 发送失败:连接已断开,请 /connect 重连", level="error") await self.notify_system("❌ 发送失败:连接已断开,请 /connect 重连", level="error")
# ── 快捷键 ──────────────────────────────────────────────────────────── # ── 快捷键 ────────────────────────────────────────────────────────────
+30 -48
View File
@@ -7,6 +7,11 @@
- reasoning_end 思考结束信号 - reasoning_end 思考结束信号
- complete 收尾信号:chunk 流式是内容权威来源,complete 仅用于收尾; - complete 收尾信号:chunk 流式是内容权威来源,complete 仅用于收尾;
仅当流式完全丢帧(未建 widget 且无内容)时,用 full_text 补建 仅当流式完全丢帧(未建 widget 且无内容)时,用 full_text 补建
顺序保证:不预创建空AI widget,按消息到达顺序自然挂载。
- 先到 reasoning_start → 思考区域挂在最后
- 后到 chunk → AI回答挂在思考区域之后
- 无思考直接 chunk → AI回答挂在最后
""" """
from textual.containers import VerticalScroll from textual.containers import VerticalScroll
from textual.widgets import Markdown, Static from textual.widgets import Markdown, Static
@@ -15,7 +20,8 @@ from textual.widgets import Markdown, Static
class ChatView(VerticalScroll): class ChatView(VerticalScroll):
"""可滚动聊天区。AI 回复用单个 Markdown 部件增量 append,实现流式渲染。 """可滚动聊天区。AI 回复用单个 Markdown 部件增量 append,实现流式渲染。
顺序保证:思考区域始终在 AI 正式回答之前。 不预创建空widget:避免重排复杂性,消息按接收顺序自然挂载,
天然保证思考过程在正式回答之前。
滚动策略:每条新消息挂载后调用 anchor() 锚定底部——之后内容增长 滚动策略:每条新消息挂载后调用 anchor() 锚定底部——之后内容增长
(流式 append)容器会自动保持吸底;用户上翻时 Textual 自动脱锚, (流式 append)容器会自动保持吸底;用户上翻时 Textual 自动脱锚,
下一条新消息再重新锚定。 下一条新消息再重新锚定。
@@ -28,7 +34,6 @@ class ChatView(VerticalScroll):
self._ai_text = "" self._ai_text = ""
self._reasoning_text = "" self._reasoning_text = ""
self._ai_finalized = True # 初始无活跃 AI 消息 self._ai_finalized = True # 初始无活跃 AI 消息
self._ai_prefab = False # AI widget是预创建的空widget(可重排)
@property @property
def ai_text(self) -> str: def ai_text(self) -> str:
@@ -47,47 +52,23 @@ class ChatView(VerticalScroll):
self._anchor_bottom() self._anchor_bottom()
async def start_ai(self): async def start_ai(self):
"""预建一条 AI 消息(发送用户消息后即可调用,给出即时反馈)。 """开始AI回复(标记状态,不预创建空widget)。
注意:此时只创建空widget但不显示内容,如果后续先收到思考内容, 为保持顺序正确,空widget不在此处创建。第一个chunk到来时才创建AI widget,
会在start_reasoning()中重新排列顺序,确保思考在前。 此时思考widget(如果有)已经先挂载,顺序自然正确。
""" """
if self._ai_widget is not None: if not self._ai_finalized:
return return
self._ai_text = "" self._ai_text = ""
self._ai_finalized = False self._ai_finalized = False
self._ai_prefab = True
self._ai_widget = Markdown("", classes="msg-ai")
await self.mount(self._ai_widget)
self._anchor_bottom()
async def start_reasoning(self, hint: str = "🤔 思考中..."): async def start_reasoning(self, hint: str = "🤔 思考中..."):
"""开始展示思考过程区域。 """开始展示思考过程区域。直接挂载在当前末尾位置。"""
如果AI widget已经预创建(空的),先移除它,等思考widget挂载后
再重新挂载AI widget,保证思考在回答之前。
"""
if self._reasoning_widget is not None: if self._reasoning_widget is not None:
return return
# 如果AI widget已经存在且是空的预创建状态,重新排列顺序
need_remount_ai = False
ai_saved_text = self._ai_text
if self._ai_widget is not None and self._ai_prefab:
await self._ai_widget.remove()
self._ai_widget = None
need_remount_ai = True
# 先挂载思考widget
self._reasoning_text = "" self._reasoning_text = ""
self._reasoning_widget = Static(hint + "\n", classes="msg-reasoning", markup=False) self._reasoning_widget = Static(hint + "\n", classes="msg-reasoning", markup=False)
await self.mount(self._reasoning_widget) await self.mount(self._reasoning_widget)
# 如果之前有AI内容(理论上预创建时不会有,但保险处理),重新挂载AI widget
if need_remount_ai:
self._ai_widget = Markdown(ai_saved_text, classes="msg-ai")
await self.mount(self._ai_widget)
self._anchor_bottom() self._anchor_bottom()
async def append_reasoning(self, delta: str): async def append_reasoning(self, delta: str):
@@ -113,7 +94,13 @@ class ChatView(VerticalScroll):
display = "...(前略)\n" + display[-3000:] display = "...(前略)\n" + display[-3000:]
self._reasoning_widget.update(f"💭 思考过程({len(self._reasoning_text)}字符)\n{display}\n{'─' * 30}") self._reasoning_widget.update(f"💭 思考过程({len(self._reasoning_text)}字符)\n{display}\n{'─' * 30}")
# 思考区域保留显示,不删除 # 思考区域保留显示,不删除
self._ai_prefab = False # AI widget顺序已固定,不再重排
async def _ensure_ai_widget(self):
"""确保AI widget存在(第一次chunk到来时创建)。"""
if self._ai_widget is None:
self._ai_widget = Markdown("", classes="msg-ai")
await self.mount(self._ai_widget)
self._anchor_bottom()
async def append_ai(self, delta: str): async def append_ai(self, delta: str):
"""追加增量片段(服务端 chunk 为增量语义,直接拼接)。""" """追加增量片段(服务端 chunk 为增量语义,直接拼接)。"""
@@ -122,13 +109,8 @@ class ChatView(VerticalScroll):
if self._ai_finalized: if self._ai_finalized:
# 收尾后收到的延迟 chunk,丢弃以避免在上一条回复后追加脏数据 # 收尾后收到的延迟 chunk,丢弃以避免在上一条回复后追加脏数据
return return
if self._ai_widget is None: # 第一次收到chunk时创建widget(此时思考widget如果存在已经在前面了)
# AI widget还没创建,先创建 await self._ensure_ai_widget()
self._ai_text = ""
self._ai_prefab = False
self._ai_widget = Markdown("", classes="msg-ai")
await self.mount(self._ai_widget)
self._ai_prefab = False # 开始有实际内容了
self._ai_widget.append(delta) self._ai_widget.append(delta)
self._ai_text += delta self._ai_text += delta
# anchor 状态下容器会自动吸底;保险起见刷新后再滚一次 # anchor 状态下容器会自动吸底;保险起见刷新后再滚一次
@@ -140,9 +122,9 @@ class ChatView(VerticalScroll):
chunk 流式追加是内容权威来源,complete 只做收尾: chunk 流式追加是内容权威来源,complete 只做收尾:
- 已收尾则直接返回(防止 dialogue_ended + complete 双发导致重复挂载) - 已收尾则直接返回(防止 dialogue_ended + complete 双发导致重复挂载)
- widget 存在且有流式内容:保持已渲染内容不变,仅清空引用 - widget 存在且有流式内容:保持已渲染内容不变,仅清空引用
- widget 存在但无内容(预建空框)且有 full_text:用 full_text 填充 - widget 存在但无内容(异常场景)且有 full_text:用 full_text 填充
- widget 存在但无内容也无 full_text:移除空框 - widget 不存在(未收到任何 chunk)且有 full_text:补建
- widget 不存在(流式丢帧/start_ai 未调用)且有 full_text:补建 - widget 不存在且无内容:正常无内容场景
""" """
if self._ai_finalized: if self._ai_finalized:
return return
@@ -154,24 +136,25 @@ class ChatView(VerticalScroll):
# 有流式内容,chunk 已完整渲染,直接保留,不做替换 # 有流式内容,chunk 已完整渲染,直接保留,不做替换
pass pass
elif full_text: elif full_text:
# 预建了空 widget 但没收到任何 chunk:用 full_text 填充 # widget存在但没收到任何chunk(异常):用full_text填充
w.update(full_text) w.update(full_text)
self._ai_text = full_text self._ai_text = full_text
else: else:
# 预建后没有任何输出:移除空框 # widget存在但没有任何输出:移除空框(正常不会走到这里,因为不预创建)
await w.remove() await w.remove()
else: else:
# widget 不存在(start_ai 未调用),有 full_text 则补建 # widget不存在(start_ai后没收到任何chunk),有full_text则补建
if full_text: if full_text:
self._ai_text = full_text self._ai_text = full_text
await self.mount(Markdown(full_text, classes="msg-ai")) self._ai_widget = Markdown(full_text, classes="msg-ai")
await self.mount(self._ai_widget)
self._ai_widget = None
self._anchor_bottom() self._anchor_bottom()
self._ai_text = "" self._ai_text = ""
self._reasoning_widget = None self._reasoning_widget = None
self._reasoning_text = "" self._reasoning_text = ""
self._ai_finalized = True self._ai_finalized = True
self._ai_prefab = False
self.call_after_refresh(self.scroll_end, animate=False) self.call_after_refresh(self.scroll_end, animate=False)
async def reset(self): async def reset(self):
@@ -181,5 +164,4 @@ class ChatView(VerticalScroll):
self._ai_text = "" self._ai_text = ""
self._reasoning_text = "" self._reasoning_text = ""
self._ai_finalized = True self._ai_finalized = True
self._ai_prefab = False
await self.remove_children() await self.remove_children()
+1 -1
View File
@@ -7,7 +7,7 @@ INCLU_SDIR := inc
CPP := g++ CPP := g++
INCS := -I inc -I/opt/homebrew/include INCS := -I inc -I/opt/homebrew/include
CPPFLAGS := --std=c++20 -O3 -pipe -Wall -Werror CPPFLAGS := --std=c++20 -O3 -pipe -Wall -Werror
LIBS := -L/opt/homebrew/lib -lssl -lcrypto -lpthread LIBS := -L/opt/homebrew/lib -lssl -lcrypto -lcurl -lpthread
BUILD_TDIR := build BUILD_TDIR := build
$(shell mkdir -p $(BUILD_TDIR)) $(shell mkdir -p $(BUILD_TDIR))
+231
View File
@@ -0,0 +1,231 @@
# libhttp - UniServe HTTP/WebSocket Library
C++20 network library for the UniServe microservice framework, providing unified HTTP/HTTPS server and HTTP/WebSocket client capabilities built on Boost.Beast and libcurl.
## Features
### HttpServer
- **HTTP/HTTPS Server**: Asynchronous HTTP 1.1 server with TLS support
- **WebSocket Integration**: Single port can serve both HTTP(S) and WebSocket(S) endpoints
- **Routing**: Exact path matching + wildcard (`*`) routing
- **Middleware**: `use()` for middleware chains (CORS, auth, logging, etc.)
- **Static Files**: `setStaticDirectory()` for serving static files
- **Thread Pool**: Multi-threaded I/O with `asio::strand` for thread safety
- **Client Management**: WebSocket client tracking, broadcast, and targeted send
### HttpClient
- **HTTP Client**: Synchronous HTTP requests via libcurl (GET/POST/PUT/DELETE/PATCH/HEAD/OPTIONS)
- **WebSocket Client**: Synchronous WebSocket client via Boost.Beast
- **Dual Protocol**: Same client instance handles both HTTP and WebSocket
- **ws:// and wss://**: Supports both plain and TLS-encrypted WebSocket connections
- **Binary Support**: Text and binary message sending
- **Backward Compatible**: Original API preserved; `WebSocketClient`/`WebSocketClientBase` are aliases for `HttpClient`
## Dependencies
- C++20 compiler
- Boost.Beast / Boost.Asio (header-only)
- OpenSSL (for TLS)
- libcurl (for HTTP client)
- pthread
## Build
```bash
make # builds build/libus_http.so
make install # installs to /usr/lib and /usr/include/us (requires root)
```
Makefile links: `-lssl -lcrypto -lcurl -lpthread` with Homebrew paths for macOS.
## Quick Start
### HTTP + WebSocket Server
```cpp
#include <us/HttpServer.hpp>
using namespace pmc::net;
int main() {
// HTTP server on port 8080
HttpServer server("0.0.0.0", 8080, 4); // 4 threads
// HTTP GET route
server.get("/hello", [](const auto& req, const auto& params) {
http::response<http::string_body> res;
res.result(http::status::ok);
res.set(http::field::content_type, "text/plain");
res.body() = "Hello, World!";
res.prepare_payload();
return res;
});
// WebSocket route
server.ws("/ws", [](const std::string& clientId,
const std::string& message,
std::function<void(const std::string&)> send) {
send("Echo: " + message);
});
server.onWsConnect([](const std::string& id, const std::string& addr) {
std::cout << "Client " << id << " connected from " << addr << std::endl;
});
server.onWsDisconnect([](const std::string& id) {
std::cout << "Client " << id << " disconnected" << std::endl;
});
server.start();
// server.run(); // blocks with ioc.run()
return 0;
}
```
### HTTP + WebSocket Client
```cpp
#include <us/HttpClient.hpp>
using namespace pmc::net;
// HTTP request
HttpClient http;
http.set_header("Content-Type: application/json");
http.post_stream_sync("http://localhost:8080/api/data", "{}",
[](const std::string& resp) {
std::cout << "Response: " << resp << std::endl;
});
// WebSocket connection
HttpClient ws;
if (ws.ws_connect("ws://localhost:8080/ws")) {
ws.ws_send("Hello WebSocket!");
std::string reply = ws.ws_recv();
std::cout << "Got: " << reply << std::endl;
ws.ws_close();
}
// WSS (TLS) connection
HttpClient wss;
wss.ws_connect("wss://example.com/socket");
```
### HTTPS Server
```cpp
HttpServer secure("0.0.0.0", 443, "/path/to/cert.pem", "/path/to/key.pem", 4);
secure.get("/", handler);
secure.start();
```
## API Reference
### HttpServer
**Constructors**
```cpp
HttpServer(const std::string& address, unsigned short port, unsigned int threads = 1);
HttpServer(const std::string& address, unsigned short port,
const std::string& cert_file, const std::string& key_file,
unsigned int threads = 1);
```
**HTTP Routes**
```cpp
void get(const std::string& path, HttpRequestHandler handler);
void post(const std::string& path, HttpRequestHandler handler);
void put(const std::string& path, HttpRequestHandler handler);
void del(const std::string& path, HttpRequestHandler handler);
void options(const std::string& path, HttpRequestHandler handler);
void use(MiddlewareHandler middleware);
void setStaticDirectory(const std::string& path);
```
**WebSocket Routes**
```cpp
void ws(const std::string& path, WebSocketMessageHandler handler);
void onWsConnect(WebSocketConnectHandler handler);
void onWsDisconnect(WebSocketDisconnectHandler handler);
bool sendToWsClient(const std::string& clientId, const std::string& message);
size_t broadcastWs(const std::string& message);
void disconnectWsClient(const std::string& clientId);
size_t getWsClientCount() const;
```
**Lifecycle**
```cpp
void start(); // non-blocking, spawns worker threads
void run(); // blocking (calls start() then ioc.run())
void stop();
bool isRunning() const;
```
### HttpClient
**HTTP Methods**
```cpp
void set_header(const std::string& header);
void clear_headers();
const std::map<std::string, std::string>& get_response_headers() const;
long get_response_code() const;
void get_stream_sync(const std::string& url, ResponseCallback cb);
void post_stream_sync(const std::string& url, const std::string& data, ResponseCallback cb);
void put_stream_sync(const std::string& url, const std::string& data, ResponseCallback cb);
void delete_stream_sync(const std::string& url, ResponseCallback cb);
void request_stream_sync(HttpMethod method, const std::string& url,
const std::string& data, ResponseCallback cb);
```
**WebSocket Methods**
```cpp
bool ws_connect(const std::string& url, int timeout_ms = 5000);
bool ws_connect(const std::string& host, const std::string& port,
const std::string& path = "/", bool ssl = false, int timeout_ms = 5000);
bool ws_send(const std::string& message);
bool ws_send_binary(const void* data, size_t size);
std::string ws_recv(int timeout_ms = 0);
bool ws_is_connected() const;
void ws_close();
std::string ws_get_host() const;
std::string ws_get_port() const;
std::string ws_get_path() const;
```
### Type Definitions
```cpp
using HttpRequestHandler = std::function<http::response<http::string_body>(
const http::request<http::string_body>&,
const std::unordered_map<std::string, std::string>&)>;
using MiddlewareHandler = std::function<bool(
const http::request<http::string_body>&,
http::response<http::string_body>&,
const std::unordered_map<std::string, std::string>&)>;
using WebSocketMessageHandler = std::function<void(
const std::string& clientId,
const std::string& message,
std::function<void(const std::string&)> sendCallback)>;
using WebSocketConnectHandler = std::function<void(
const std::string& clientId,
const std::string& remoteAddress)>;
using WebSocketDisconnectHandler = std::function<void(
const std::string& clientId)>;
```
## Backward Compatibility
The following type aliases are provided for existing code that used the standalone WebSocket client:
```cpp
using WebSocketClient = HttpClient;
using WebSocketClientBase = HttpClient;
```
## Version History
- **v3.0.0**: Merged WebSocketServer into HttpServer and WebSocketClient into HttpClient. Single-port HTTP+WebSocket support.
- **v2.x**: Separate HttpServer, WebSocketServer, and HttpClient modules.
+272 -19
View File
@@ -2,10 +2,29 @@
#include <string> #include <string>
#include <functional> #include <functional>
#include <map> #include <map>
#include <memory>
#include <stdexcept>
#include <chrono>
#include <iostream>
#include <curl/curl.h> #include <curl/curl.h>
#include <openssl/ssl.h>
// Boost.Beast for WebSocket client
#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/asio/ssl/stream.hpp>
namespace pmc::net { namespace pmc::net {
namespace beast = boost::beast;
namespace asio = boost::asio;
namespace websocket = beast::websocket;
using tcp = asio::ip::tcp;
enum class HttpMethod { enum class HttpMethod {
GET, GET,
POST, POST,
@@ -26,6 +45,7 @@ public:
} }
~HttpClient() { ~HttpClient() {
ws_close();
if (curl_) { if (curl_) {
curl_easy_cleanup(curl_); curl_easy_cleanup(curl_);
} }
@@ -33,6 +53,9 @@ public:
curl_slist_free_all(headers_); curl_slist_free_all(headers_);
} }
} }
HttpClient(const HttpClient&) = delete;
HttpClient& operator=(const HttpClient&) = delete;
void set_header(const std::string& header) { void set_header(const std::string& header) {
headers_ = curl_slist_append(headers_, header.c_str()); headers_ = curl_slist_append(headers_, header.c_str());
@@ -53,7 +76,7 @@ public:
return response_code_; return response_code_;
} }
// ===== 修复:直接转发,不按行处理 ===== // ===== HTTP Methods (Synchronous, libcurl-based) =====
void request_stream_sync( void request_stream_sync(
HttpMethod method, HttpMethod method,
const std::string& url, const std::string& url,
@@ -74,7 +97,6 @@ public:
curl_easy_setopt(curl_, CURLOPT_HEADERFUNCTION, header_callback); curl_easy_setopt(curl_, CURLOPT_HEADERFUNCTION, header_callback);
curl_easy_setopt(curl_, CURLOPT_HEADERDATA, this); curl_easy_setopt(curl_, CURLOPT_HEADERDATA, this);
// ===== 关键修复:直接收集原始数据,不按行处理 =====
std::string full_response; std::string full_response;
curl_easy_setopt(curl_, CURLOPT_WRITEFUNCTION, write_callback); curl_easy_setopt(curl_, CURLOPT_WRITEFUNCTION, write_callback);
curl_easy_setopt(curl_, CURLOPT_WRITEDATA, &full_response); curl_easy_setopt(curl_, CURLOPT_WRITEDATA, &full_response);
@@ -90,7 +112,6 @@ public:
curl_easy_getinfo(curl_, CURLINFO_RESPONSE_CODE, &response_code_); curl_easy_getinfo(curl_, CURLINFO_RESPONSE_CODE, &response_code_);
// ===== 完整转发所有数据(包括换行符) =====
if (!full_response.empty()) { if (!full_response.empty()) {
callback(full_response); callback(full_response);
} }
@@ -112,18 +133,239 @@ public:
request_stream_sync(HttpMethod::DELETE, url, "", callback); request_stream_sync(HttpMethod::DELETE, url, "", callback);
} }
// ===== WebSocket Methods (Synchronous, Boost.Beast-based) =====
/**
* Connect to WebSocket server (ws:// or wss://)
* @param url Full URL like "ws://host:port/path" or "wss://host:port/path"
* @param timeout_ms Connection timeout in milliseconds
* @return true on success
*/
bool ws_connect(const std::string& url, int timeout_ms = 5000) {
std::string host, port, path;
bool ssl = false;
if (url.size() >= 6 && url.substr(0, 6) == "wss://") {
ssl = true;
parse_ws_url(url.substr(6), host, port, path, "443");
} else if (url.size() >= 5 && url.substr(0, 5) == "ws://") {
ssl = false;
parse_ws_url(url.substr(5), host, port, path, "80");
} else {
return false;
}
return ws_connect(host, port, path, ssl, timeout_ms);
}
/**
* Connect to WebSocket server with explicit parameters
*/
bool ws_connect(const std::string& host, const std::string& port,
const std::string& path = "/", bool ssl = false, int timeout_ms = 5000) {
ws_close();
ws_state_ = std::make_unique<WsState>();
ws_state_->ssl_enabled = ssl;
ws_state_->host = host;
ws_state_->port = port;
ws_state_->path = path.empty() ? "/" : path;
try {
tcp::resolver resolver(ws_state_->ioc);
auto const results = resolver.resolve(host, port);
if (ssl) {
// wss:// path: tcp::socket → asio::ssl::stream → websocket::stream
ws_state_->ssl_ctx = std::make_unique<asio::ssl::context>(asio::ssl::context::tlsv12_client);
ws_state_->ssl_ctx->set_verify_mode(asio::ssl::verify_none);
// Create socket wrapped in beast::tcp_stream for timeout support
ws_state_->tcp_stream = std::make_unique<beast::tcp_stream>(ws_state_->ioc);
// Set connect timeout
ws_state_->tcp_stream->expires_after(std::chrono::milliseconds(timeout_ms));
asio::connect(ws_state_->tcp_stream->socket(), results);
// Now build SSL stream on top of the socket
ws_state_->ssl_stream = std::make_unique<asio::ssl::stream<tcp::socket&>>(
ws_state_->tcp_stream->socket(), *ws_state_->ssl_ctx);
// Set SNI
if (!SSL_set_tlsext_host_name(ws_state_->ssl_stream->native_handle(), host.c_str())) {
throw std::runtime_error("Failed to set SNI hostname");
}
// SSL handshake
ws_state_->ssl_stream->handshake(asio::ssl::stream_base::client);
// Build websocket stream on top of SSL stream
ws_state_->wss_stream = std::make_unique<websocket::stream<asio::ssl::stream<tcp::socket&>&>>(
*ws_state_->ssl_stream);
// WebSocket handshake
ws_state_->tcp_stream->expires_never();
ws_state_->wss_stream->set_option(websocket::stream_base::timeout::suggested(beast::role_type::client));
ws_state_->wss_stream->set_option(websocket::stream_base::decorator(
[](websocket::request_type& req) {
req.set(beast::http::field::user_agent, "UniServe-HttpClient");
}));
ws_state_->wss_stream->handshake(host, ws_state_->path);
} else {
// ws:// path: tcp::socket wrapped in beast::tcp_stream → websocket::stream
ws_state_->tcp_stream = std::make_unique<beast::tcp_stream>(ws_state_->ioc);
ws_state_->tcp_stream->expires_after(std::chrono::milliseconds(timeout_ms));
asio::connect(ws_state_->tcp_stream->socket(), results);
// Build websocket stream on top of tcp_stream's socket (by reference)
ws_state_->ws_stream = std::make_unique<websocket::stream<tcp::socket&>>(
ws_state_->tcp_stream->socket());
ws_state_->tcp_stream->expires_never();
ws_state_->ws_stream->set_option(websocket::stream_base::timeout::suggested(beast::role_type::client));
ws_state_->ws_stream->set_option(websocket::stream_base::decorator(
[](websocket::request_type& req) {
req.set(beast::http::field::user_agent, "UniServe-HttpClient");
}));
ws_state_->ws_stream->handshake(host, ws_state_->path);
}
ws_state_->connected = true;
return true;
} catch (const std::exception& e) {
std::cerr << "[HttpClient::ws_connect] " << e.what() << std::endl;
ws_state_.reset();
return false;
}
}
/**
* Send a text message over WebSocket
*/
bool ws_send(const std::string& message) {
if (!ws_state_ || !ws_state_->connected) return false;
try {
if (ws_state_->ssl_enabled && ws_state_->wss_stream) {
ws_state_->wss_stream->write(asio::buffer(message));
} else if (ws_state_->ws_stream) {
ws_state_->ws_stream->write(asio::buffer(message));
}
return true;
} catch (const std::exception& e) {
std::cerr << "[HttpClient::ws_send] " << e.what() << std::endl;
ws_state_->connected = false;
return false;
}
}
/**
* Send binary data over WebSocket
*/
bool ws_send_binary(const void* data, size_t size) {
if (!ws_state_ || !ws_state_->connected) return false;
try {
if (ws_state_->ssl_enabled && ws_state_->wss_stream) {
ws_state_->wss_stream->binary(true);
ws_state_->wss_stream->write(asio::buffer(data, size));
ws_state_->wss_stream->text(true);
} else if (ws_state_->ws_stream) {
ws_state_->ws_stream->binary(true);
ws_state_->ws_stream->write(asio::buffer(data, size));
ws_state_->ws_stream->text(true);
}
return true;
} catch (const std::exception& e) {
std::cerr << "[HttpClient::ws_send_binary] " << e.what() << std::endl;
ws_state_->connected = false;
return false;
}
}
/**
* Receive a message from WebSocket (blocking)
* @param timeout_ms Timeout in milliseconds (0 = wait forever)
* @return Received message, or empty string on error/timeout/close
*/
std::string ws_recv(int timeout_ms = 0) {
if (!ws_state_ || !ws_state_->connected) return "";
try {
beast::flat_buffer buffer;
// For synchronous recv with timeout, we use websocket stream's own timeout options
// since we set suggested timeouts during handshake
if (ws_state_->ssl_enabled && ws_state_->wss_stream) {
ws_state_->wss_stream->read(buffer);
} else if (ws_state_->ws_stream) {
ws_state_->ws_stream->read(buffer);
}
return beast::buffers_to_string(buffer.data());
} catch (const beast::system_error& e) {
if (e.code() == websocket::error::closed) {
ws_state_->connected = false;
} else {
std::cerr << "[HttpClient::ws_recv] " << e.what() << std::endl;
ws_state_->connected = false;
}
return "";
} catch (const std::exception& e) {
std::cerr << "[HttpClient::ws_recv] " << e.what() << std::endl;
ws_state_->connected = false;
return "";
}
}
bool ws_is_connected() const {
return ws_state_ && ws_state_->connected;
}
void ws_close() {
if (!ws_state_) return;
try {
if (ws_state_->connected) {
beast::error_code ec;
if (ws_state_->ssl_enabled && ws_state_->wss_stream) {
ws_state_->wss_stream->close(websocket::close_code::normal, ec);
} else if (ws_state_->ws_stream) {
ws_state_->ws_stream->close(websocket::close_code::normal, ec);
}
}
} catch (...) {
// Ignore close errors
}
ws_state_.reset();
}
std::string ws_get_host() const { return ws_state_ ? ws_state_->host : ""; }
std::string ws_get_port() const { return ws_state_ ? ws_state_->port : ""; }
std::string ws_get_path() const { return ws_state_ ? ws_state_->path : ""; }
private: private:
CURL* curl_ = nullptr; CURL* curl_ = nullptr;
struct curl_slist* headers_ = nullptr; struct curl_slist* headers_ = nullptr;
std::map<std::string, std::string> response_headers_; std::map<std::string, std::string> response_headers_;
long response_code_ = 0; long response_code_ = 0;
// WebSocket state - layered streams with reference ownership
struct WsState {
asio::io_context ioc;
std::unique_ptr<beast::tcp_stream> tcp_stream; // owns the socket
std::unique_ptr<asio::ssl::context> ssl_ctx; // SSL context
std::unique_ptr<asio::ssl::stream<tcp::socket&>> ssl_stream; // references tcp_stream->socket()
std::unique_ptr<websocket::stream<tcp::socket&>> ws_stream; // ws:// references tcp_stream->socket()
std::unique_ptr<websocket::stream<asio::ssl::stream<tcp::socket&>&>> wss_stream; // wss:// references ssl_stream
bool ssl_enabled = false;
bool connected = false;
std::string host;
std::string port;
std::string path;
};
std::unique_ptr<WsState> ws_state_;
void set_http_method(HttpMethod method, const std::string& data) { void set_http_method(HttpMethod method, const std::string& data) {
switch (method) { switch (method) {
case HttpMethod::GET: case HttpMethod::GET:
curl_easy_setopt(curl_, CURLOPT_HTTPGET, 1L); curl_easy_setopt(curl_, CURLOPT_HTTPGET, 1L);
break; break;
case HttpMethod::POST: case HttpMethod::POST:
curl_easy_setopt(curl_, CURLOPT_POST, 1L); curl_easy_setopt(curl_, CURLOPT_POST, 1L);
if (!data.empty()) { if (!data.empty()) {
@@ -131,7 +373,6 @@ private:
curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, data.size()); curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, data.size());
} }
break; break;
case HttpMethod::PUT: case HttpMethod::PUT:
curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "PUT"); curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "PUT");
if (!data.empty()) { if (!data.empty()) {
@@ -139,11 +380,9 @@ private:
curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, data.size()); curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, data.size());
} }
break; break;
case HttpMethod::DELETE: case HttpMethod::DELETE:
curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "DELETE"); curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "DELETE");
break; break;
case HttpMethod::PATCH: case HttpMethod::PATCH:
curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "PATCH"); curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "PATCH");
if (!data.empty()) { if (!data.empty()) {
@@ -151,50 +390,64 @@ private:
curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, data.size()); curl_easy_setopt(curl_, CURLOPT_POSTFIELDSIZE, data.size());
} }
break; break;
case HttpMethod::HEAD: case HttpMethod::HEAD:
curl_easy_setopt(curl_, CURLOPT_NOBODY, 1L); curl_easy_setopt(curl_, CURLOPT_NOBODY, 1L);
break; break;
case HttpMethod::OPTIONS: case HttpMethod::OPTIONS:
curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "OPTIONS"); curl_easy_setopt(curl_, CURLOPT_CUSTOMREQUEST, "OPTIONS");
break; break;
} }
} }
static size_t header_callback(char* buffer, size_t size, size_t nitems, void* userdata) { static size_t header_callback(char* buffer, size_t size, size_t nitems, void* userdata) {
size_t total_size = size * nitems; size_t total_size = size * nitems;
HttpClient* client = static_cast<HttpClient*>(userdata); HttpClient* client = static_cast<HttpClient*>(userdata);
std::string header_line(buffer, total_size); std::string header_line(buffer, total_size);
size_t colon_pos = header_line.find(':'); size_t colon_pos = header_line.find(':');
if (colon_pos != std::string::npos) { if (colon_pos != std::string::npos) {
std::string key = header_line.substr(0, colon_pos); std::string key = header_line.substr(0, colon_pos);
std::string value = header_line.substr(colon_pos + 1); std::string value = header_line.substr(colon_pos + 1);
key.erase(0, key.find_first_not_of(" \t")); key.erase(0, key.find_first_not_of(" \t"));
key.erase(key.find_last_not_of(" \t") + 1); key.erase(key.find_last_not_of(" \t") + 1);
value.erase(0, value.find_first_not_of(" \t")); value.erase(0, value.find_first_not_of(" \t"));
value.erase(value.find_last_not_of(" \t\r\n") + 1); value.erase(value.find_last_not_of(" \t\r\n") + 1);
if (!key.empty() && !value.empty()) { if (!key.empty() && !value.empty()) {
client->response_headers_[key] = value; client->response_headers_[key] = value;
} }
} }
return total_size; return total_size;
} }
// ===== 修复:直接写入原始数据,不按行处理 =====
static size_t write_callback(char* ptr, size_t size, size_t nmemb, void* userdata) { static size_t write_callback(char* ptr, size_t size, size_t nmemb, void* userdata) {
size_t total_size = size * nmemb; size_t total_size = size * nmemb;
std::string* response = static_cast<std::string*>(userdata); std::string* response = static_cast<std::string*>(userdata);
// 直接追加原始数据,保留所有换行符和特殊字符
response->append(ptr, total_size); response->append(ptr, total_size);
return total_size; return total_size;
} }
static void parse_ws_url(const std::string& url_part, std::string& host, std::string& port, std::string& path, const char* default_port) {
size_t path_pos = url_part.find('/');
std::string host_port;
if (path_pos != std::string::npos) {
host_port = url_part.substr(0, path_pos);
path = url_part.substr(path_pos);
} else {
host_port = url_part;
path = "/";
}
size_t port_pos = host_port.find(':');
if (port_pos != std::string::npos) {
host = host_port.substr(0, port_pos);
port = host_port.substr(port_pos + 1);
} else {
host = host_port;
port = default_port;
}
}
}; };
// Backward compatibility aliases
using WebSocketClientBase = HttpClient;
using WebSocketClient = HttpClient;
} // namespace pmc::net } // namespace pmc::net