如何在C++/MQL4中原生监听Server-Sent Event(SSE)？

核心原理 #

SSE本质是HTTP长连接下的文本流传输，服务器会持续推送格式规范的文本数据。不需要依赖专门的库，只要能发起HTTP请求并保持连接、逐行解析响应流就能实现。

MQL4实现步骤 #

MQL4自带WebRequest函数，但默认是短连接，要实现长连接需要手动处理：

1. 发起HTTP请求，设置长连接头部 #

必须设置Connection: keep-alive和Accept: text/event-stream两个关键头部，告知服务器要建立SSE连接：

void OnStart() {
    string url = "https://sdk.m.pipedream.net/pipelines/p_jmClnMy/sse";
    string headers = "Connection: keep-alive\r\nAccept: text/event-stream\r\n";
    char postData[];
    char response[];
    int timeout = 0; // 设置为0保持长连接，避免超时断开
    
    // 发起请求并保持连接
    int ret = WebRequest("GET", url, headers, timeout, postData, 0, response, 0);
    
    if(ret == 200) {
        // 持续读取响应流
        ReadSSEStream(response);
    } else {
        Print("请求失败，错误码: ", ret);
    }
}

2. 解析SSE流数据 #

SSE消息格式固定，每条消息由多行组成，核心字段包括：

• event: [事件名]：对应JS中的监听事件（比如你用的events）
• data: [内容]：消息主体，可能跨多行
• 空行表示单条消息结束

实现解析函数：

void ReadSSEStream(char &stream[]) {
    string currentEvent = "";
    string currentData = "";
    
    while(true) {
        // 逐行读取流数据
        string line = ReadLineFromStream(stream);
        
        if(line == "") {
            // 空行触发消息处理
            if(currentEvent == "events" && currentData != "") {
                Print("收到SSE事件: ", currentData);
            }
            // 重置临时变量
            currentEvent = "";
            currentData = "";
            continue;
        }
        
        // 解析行内容
        if(StringFind(line, "event: ") == 0) {
            currentEvent = StringSubstr(line, 7);
        } else if(StringFind(line, "data: ") == 0) {
            currentData += StringSubstr(line, 6) + "\n";
        }
        // 可按需扩展处理id字段
    }
}

// 辅助函数：从字符流中读取一行
string ReadLineFromStream(char &stream[]) {
    static int pos = 0;
    string line = "";
    
    while(pos < ArraySize(stream)) {
        if(stream[pos] == '\n') {
            pos++;
            // 去除可能的换行符\r
            if(line != "" && StringGetChar(line, StringLen(line)-1) == '\r') {
                line = StringSubstr(line, 0, StringLen(line)-1);
            }
            return line;
        }
        line += CharToStr(stream[pos]);
        pos++;
    }
    return "";
}

注意事项 #

• 部分MT4版本可能会强制断开超时的长连接，需要测试调整超时参数
• 要添加连接断开后的重连逻辑，避免丢失消息
• 需动态处理流数据缓冲区，防止内存溢出

C++实现思路 #

C++中可以用libcurl库实现长连接监听：

1. 初始化libcurl，设置目标SSE地址
2. 添加Accept: text/event-stream和Connection: keep-alive请求头部
3. 设置回调函数实时接收服务器推送的数据
4. 在回调函数中按SSE格式解析数据，处理events事件

#include <curl/curl.h>
#include <string>
#include <iostream>

std::string currentEvent;
std::string currentData;

size_t SSECallback(char *ptr, size_t size, size_t nmemb, void *userdata) {
    size_t totalSize = size * nmemb;
    std::string data(ptr, totalSize);
    
    size_t pos = 0;
    while(pos < data.size()) {
        size_t newlinePos = data.find('\n', pos);
        if(newlinePos == std::string::npos) break;
        
        std::string line = data.substr(pos, newlinePos - pos);
        // 去除末尾的\r
        if(!line.empty() && line.back() == '\r') {
            line.pop_back();
        }
        
        if(line.empty()) {
            // 空行触发消息处理
            if(currentEvent == "events" && !currentData.empty()) {
                std::cout << "收到SSE事件: " << currentData << std::endl;
            }
            currentEvent.clear();
            currentData.clear();
        } else if(line.substr(0, 7) == "event: ") {
            currentEvent = line.substr(7);
        } else if(line.substr(0, 6) == "data: ") {
            currentData += line.substr(6) + "\n";
        }
        
        pos = newlinePos + 1;
    }
    
    return totalSize;
}

int main() {
    CURL *curl = curl_easy_init();
    if(curl) {
        struct curl_slist *headers = NULL;
        headers = curl_slist_append(headers, "Accept: text/event-stream");
        headers = curl_slist_append(headers, "Connection: keep-alive");
        
        curl_easy_setopt(curl, CURLOPT_URL, "https://sdk.m.pipedream.net/pipelines/p_jmClnMy/sse");
        curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, SSECallback);
        curl_easy_setopt(curl, CURLOPT_TIMEOUT, 0L); // 无限超时
        
        CURLcode res = curl_easy_perform(curl);
        if(res != CURLE_OK) {
            std::cerr << "请求失败: " << curl_easy_strerror(res) << std::endl;
        }
        
        curl_slist_free_all(headers);
        curl_easy_cleanup(curl);
    }
    return 0;
}

内容的提问来源于stack exchange，提问作者Hamza L.