减少内存拷贝,提高websocket服务器性能

src/Http/HttpSession.cpp

@@ -954,12 +954,12 @@ void HttpSession::onWrite(const Buffer::Ptr &buffer) {
     header._reserved = 0;
     header._opcode = WebSocketHeader::BINARY;
     header._mask_flag = false;
-    WebSocketSplitter::encode(header,(uint8_t *)buffer->data(),buffer->size());
+    WebSocketSplitter::encode(header,buffer);
 }
 
-void HttpSession::onWebSocketEncodeData(const uint8_t *ptr,uint64_t len){
-    _ui64TotalBytes += len;
-    send((char *)ptr,len);
+void HttpSession::onWebSocketEncodeData(const Buffer::Ptr &buffer){
+    _ui64TotalBytes += buffer->size();
+    send(buffer);
 }
 
 void HttpSession::onDetach() {

src/Http/HttpSession.h

@@ -104,10 +104,9 @@ protected:
 
 	/**
     * 发送数据进行websocket协议打包后回调
-    * @param ptr
-    * @param len
+    * @param buffer
     */
-	void onWebSocketEncodeData(const uint8_t *ptr,uint64_t len) override;
+	void onWebSocketEncodeData(const Buffer::Ptr &buffer) override;
 private:
 	inline void Handle_Req_GET(int64_t &content_len);
 	inline void Handle_Req_POST(int64_t &content_len);

src/Http/WebSocketSession.h

@@ -89,7 +89,7 @@ protected:
                     header._reserved = 0;
                     header._opcode = WebSocketHeader::TEXT;
                     header._mask_flag = false;
-                    strongSelf->WebSocketSplitter::encode(header,(uint8_t *)buf->data(),buf->size());
+                    strongSelf->WebSocketSplitter::encode(header,buf);
                 }
                 return buf->size();
             });
@@ -118,12 +118,12 @@ protected:
 
         switch (header._opcode){
             case WebSocketHeader::CLOSE:{
-                HttpSessionType::encode(header,nullptr,0);
+                HttpSessionType::encode(header,nullptr);
             }
                 break;
             case WebSocketHeader::PING:{
                 const_cast<WebSocketHeader&>(header)._opcode = WebSocketHeader::PONG;
-                HttpSessionType::encode(header,(uint8_t *)_remian_data.data(),_remian_data.size());
+                HttpSessionType::encode(header,std::make_shared<BufferString>(_remian_data));
             }
                 break;
             case WebSocketHeader::CONTINUATION:{
@@ -132,8 +132,7 @@ protected:
                 break;
             case WebSocketHeader::TEXT:
             case WebSocketHeader::BINARY:{
-                BufferString::Ptr buffer = std::make_shared<BufferString>(_remian_data);
-                _session->onRecv(buffer);
+                _session->onRecv(std::make_shared<BufferString>(_remian_data));
             }
                 break;
             default:
@@ -145,11 +144,10 @@ protected:
 
     /**
     * 发送数据进行websocket协议打包后回调
-    * @param ptr
-    * @param len
+    * @param buffer
     */
-    void onWebSocketEncodeData(const uint8_t *ptr,uint64_t len) override{
-        SocketHelper::send((char *)ptr,len);
+    void onWebSocketEncodeData(const Buffer::Ptr &buffer) override{
+        SocketHelper::send(buffer);
     }
 private:
     typedef function<int(const Buffer::Ptr &buf)> onBeforeSendCB;

src/Http/WebSocketSplitter.cpp

@@ -164,9 +164,9 @@ void WebSocketSplitter::onPlayloadData(uint8_t *ptr, uint64_t len) {
     onWebSocketDecodePlayload(*this, _mask_flag ? ptr - len : ptr, len, _playload_offset);
 }
 
-void WebSocketSplitter::encode(const WebSocketHeader &header,uint8_t *data, const uint64_t len) {
+void WebSocketSplitter::encode(const WebSocketHeader &header,const Buffer::Ptr &buffer) {
     string ret;
-
+    auto len = buffer ? buffer->size() : 0;
     uint8_t byte = header._fin << 7 | ((header._reserved & 0x07) << 4) | (header._opcode & 0x0F) ;
     ret.push_back(byte);
 
@@ -195,16 +195,16 @@ void WebSocketSplitter::encode(const WebSocketHeader &header,uint8_t *data, cons
         ret.append((char *)header._mask.data(),4);
     }
 
-    onWebSocketEncodeData((uint8_t*)ret.data(),ret.size());
+    onWebSocketEncodeData(std::make_shared<BufferString>(std::move(ret)));
 
     if(len > 0){
         if(mask_flag){
-            uint8_t *ptr = data;
+            uint8_t *ptr = (uint8_t*)buffer->data();
             for(int i = 0; i < len ; ++i,++ptr){
                 *(ptr) ^= header._mask[i % 4];
             }
         }
-        onWebSocketEncodeData(data,len);
+        onWebSocketEncodeData(buffer);
     }
 
 }

src/Http/WebSocketSplitter.h

@@ -31,8 +31,10 @@
 #include <string>
 #include <vector>
 #include <memory>
-using namespace std;
+#include "Network/Buffer.h"
 
+using namespace std;
+using namespace toolkit;
 
 namespace mediakit {
 
@@ -85,12 +87,10 @@ public:
     /**
      * 编码一个数据包
      * 将触发2次onWebSocketEncodeData回调
-     * 第一次是数据头，第二次是负载数据
      * @param header 数据头
-     * @param data 负载数据
-     * @param len 负载数据长度
+     * @param buffer 负载数据
      */
-    void encode(const WebSocketHeader &header,uint8_t *data,const uint64_t len);
+    void encode(const WebSocketHeader &header,const Buffer::Ptr &buffer);
 protected:
     /**
      * 收到一个webSocket数据包包头，后续将继续触发onWebSocketDecodePlayload回调
@@ -119,7 +119,7 @@ protected:
      * @param ptr 数据指针
      * @param len 数据指针长度
      */
-    virtual void onWebSocketEncodeData(const uint8_t *ptr,uint64_t len){};
+    virtual void onWebSocketEncodeData(const Buffer::Ptr &buffer){};
 private:
     void onPlayloadData(uint8_t *data,uint64_t len);
 private: