SSE客户端C++实现（使用libcurl）

  1、SSE数据包格式
  如下所示，一条SSE消息中可以有一个或多个message，每个message由\n\n分隔，一个message也可以由一个或多个filed组成，每个filed由\n分隔，filed有data、id、event、retry四种。
  data表示消息数据，如"data:value\n"。
  id表示数据包编号，当连接断开重连的时间，客户端应该发送一个 HTTP 头，里面包含一个特殊的Last-Event-ID头信息，值为收到的末了一条数据的id。
  event表示自定义的事件类型。
  retry表示浏览器重新发起连接的隔断时间，当时间隔断到期，客户端应该重新发起连接。
  还可以有仅以冒号开头的行，表示解释。通常服务器每隔一段时间就会向浏览器发送一个解释，保持连接不中断。
  包含一条message的消息：

1. data: {"username": "bobby", "time": "02:33:48"}

复制代码

  包含两条message的消息：

1. data:this is message A
3. data:this is
4. data:message B

复制代码

  包含三条message的消息：

1. :explan textdata:this is message A
3. data:this is
4. data:message B

复制代码

2、SSE客户端实现

1. #include <thread>
2. #include <functional>
4. size_t dataCallback(void* ptr, size_t size, size_t nmemb, void* param)
5. {
6.     if (ptr == NULL || size == 0 || param == NULL)
7.         return 0;
9.     int s = size * nmemb;
10.     std::string strMsg((char*)ptr, s);
11.     CSSEClient* client = static_cast<CSSEClient*>(param);
12.     if (client && client->onMessage)
13.         client->onMessage(strMsg);
15.     return s;
16. }
18. int progressCallback(void* param, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow) {
19.     CSSEClient* client = static_cast<CSSEClient*>(param);
20.     return client->isExit() ? 1 : 0;  // 返回非0值会中断curl_easy_perform()
21. }
23. class CSSEClient
24. {
25. public:
26.     CSSEClient(const std::string& strURL) {
27.         curl_global_init(CURL_GLOBAL_DEFAULT);
28.         _strURL = strURL;
29.         _thread = std::thread(&CSSEClient::run, this);
30.     }
31.     virtual ~CSSEClient(){
32.         _bExit = true;
33.         _thread.join();
34.         curl_global_cleanup();
35.     }
36.     bool isExit() { return _bExit; }
37.     std::function<void(const std::string&)> onMessage;
38. private:
39.     void run() {
40.         while (!_bExit) {
41.             subscribeSSE();
42.             if (_bExit)
43.                 break;
44.             std::this_thread::sleep_for(std::chrono::seconds(1));
45.         }
46.     }
47.     void subscribeSSE() {
48.         CURL* curl = curl_easy_init();
49.         if (!curl) return;
51.         struct curl_slist* headers = NULL;
52.         headers = curl_slist_append(headers, "Accept: text/event-stream");
53.         headers = curl_slist_append(headers, "Cache-Control: no-cache");
54.         curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
56.         curl_easy_setopt(curl, CURLOPT_URL, _strURL.c_str());
57.         curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, dataCallback);
58.         curl_easy_setopt(curl, CURLOPT_WRITEDATA, this);
59.         curl_easy_setopt(curl, CURLOPT_TIMEOUT, 0L);
61.         curl_easy_setopt(curl, CURLOPT_TCP_KEEPALIVE, 1L); //开启TCP心跳
62.         curl_easy_setopt(curl, CURLOPT_TCP_KEEPIDLE, 60L); //空闲60秒后探测
63.         curl_easy_setopt(curl, CURLOPT_TCP_KEEPINTVL, 10L); //探测间隔10秒
65.         curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L);  // 开启传输进度（已下载/上传数据量、总数据量等）回调
66.         curl_easy_setopt(curl, CURLOPT_XFERINFOFUNCTION, progressCallback); //设置传输进度回调方法，即使没有数据传输该方法也会被调用（通常每秒1次），所以可以在该方法中判断是否需要退出
67.         curl_easy_setopt(curl, CURLOPT_XFERINFODATA, this); //设置传输回调方法的第一个参数
69.         CURLcode res = curl_easy_perform(curl);
70.         if (res == CURLE_ABORTED_BY_CALLBACK) {
71.            //SSE connection aborted by user;
72.         }
73.         else {
74.             //error
75.         }
77.         curl_easy_cleanup(curl);
78.     }
80.     std::thread _thread;
81.     std::string _strURL;
82.     std::atomic_bool _bExit = false;
83. };

复制代码

  比如，客户端订阅用户相干数据的变化：

1. CSSEClient _sseClient("http://test-api.baidu.com.cn:8080/SSE/subscribe/userID")

复制代码

  对于需要认证的SSE服务，需添加CURLOPT_USERPWD或OAuth头。