1

a8/a8.h

@@ -12,6 +12,7 @@
 #include <initializer_list>
 
 #include <a8/args.h>
+#include <a8/result.h>
 #include <a8/types.h>
 #include <a8/list.h>
 #include <a8/xvalue.h>

a8/asiotcpclient.cc

@@ -8,24 +8,25 @@
 #include <a8/a8.h>
 #include <a8/asiotcpclient.h>
 
-#ifdef USE_ASIO
+#ifdef USE_BOOST
 
 const int MAX_RECV_BUFFERSIZE = 1024 * 64;
 
 namespace a8
 {
 
-    AsioTcpClient::AsioTcpClient(asio::io_context& io_context, const std::string& remote_ip, int remote_port)
+    AsioTcpClient::AsioTcpClient(std::shared_ptr<asio::io_context> io_context, const std::string& remote_ip, int remote_port)
     {
+        io_context_ = io_context;
         remote_address_ = remote_ip;
         remote_port_ = remote_port;
         endpoint_ = std::make_shared<asio::ip::tcp::endpoint>
             (
-              asio::ip::make_address(remote_address_),
+             asio::ip::address::from_string(remote_address_),
               remote_port_
              );
         send_buffer_mutex_ = std::make_shared<std::mutex>();
-        socket_ = std::make_shared<asio::ip::tcp::socket>(io_context);
+        socket_ = std::make_shared<asio::ip::tcp::socket>(*io_context);
     }
 
     AsioTcpClient::~AsioTcpClient()

a8/asiotcpclient.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#ifdef USE_ASIO
+#ifdef USE_BOOST
 
 #include <asio.hpp>
 
@@ -16,7 +16,9 @@ namespace a8
         std::function<void (a8::AsioTcpClient*)> on_connect;
         std::function<void (a8::AsioTcpClient*)> on_disconnect;
         std::function<void (a8::AsioTcpClient*, char*, unsigned int)> on_socketread;
-        AsioTcpClient(asio::io_context& io_context, const std::string& remote_ip, int remote_port);
+        AsioTcpClient(std::shared_ptr<asio::io_context> io_context,
+                      const std::string& remote_ip,
+                      int remote_port);
         virtual ~AsioTcpClient();
         const std::string& GetRemoteAddress() { return remote_address_; }
         int GetRemotePort() { return remote_port_; }
@@ -33,6 +35,7 @@ namespace a8
         void DoSend();
 
     private:
+        std::shared_ptr<asio::io_context> io_context_;
         std::string remote_address_;
         int remote_port_ = 0;
 

a8/awaiter.cc

@@ -0,0 +1,29 @@
+#include <a8/a8.h>
+
+#include <a8/awaiter.h>
+
+namespace a8
+{
+
+    void Awaiter::Await(std::shared_ptr<Awaiter> notifyer)
+    {
+        notifyers_.push_back(notifyer);
+        DoAwait();
+    }
+
+    void Awaiter::DoDone()
+    {
+        done_ = true;
+        for (auto notifyer : notifyers_) {
+            if (!notifyer.expired()) {
+                notifyer.lock()->DoResume();
+            }
+        }
+    }
+
+    void Awaiter::SetResult(std::vector<std::any> results)
+    {
+        results_ = std::make_shared<a8::Results>(results);
+    }
+
+}

a8/awaiter.h

@@ -0,0 +1,38 @@
+#pragma once
+
+#include <a8/result.h>
+
+namespace f8
+{
+    class Coroutine;
+}
+
+namespace a8
+{
+
+    class Awaiter : public std::enable_shared_from_this<Awaiter>
+    {
+    public:
+        virtual ~Awaiter() {};
+
+        std::shared_ptr<a8::Results> GetResult() { return results_; }
+        bool Done() const { return done_; }
+        virtual void DoResume() {};
+
+    protected:
+        bool done_ = false;
+
+        std::list<std::weak_ptr<Awaiter>> notifyers_;
+        void Await(std::shared_ptr<Awaiter> notifyer);
+        virtual void DoAwait() = 0;
+        void DoDone();
+        void SetResult(std::vector<std::any> results);
+
+    private:
+        std::shared_ptr<a8::Results> results_;
+        std::function<void()> cb_;
+
+        friend class f8::Coroutine;
+    };
+
+}

a8/promise.cc

@@ -0,0 +1,7 @@
+#include <a8/a8.h>
+#include <a8/promise.h>
+
+namespace a8
+{
+
+}

a8/promise.h

@@ -0,0 +1,12 @@
+#pragma once
+
+#include <a8/awaiter.h>
+
+namespace a8
+{
+
+    class Promise : public Awaiter
+    {
+    };
+
+}

a8/result.h

@@ -0,0 +1,17 @@
+#pragma once
+
+namespace a8
+{
+    class Results
+    {
+    public:
+
+        Results(std::vector<std::any> results):results_(std::move(results)) {};
+
+        template <typename T>
+        T Get(size_t index) const { return std::any_cast<T>(results_.at(index));};
+
+    private:
+        std::vector<std::any> results_;
+    };
+}

a8/websocketclient.cc

@@ -0,0 +1,297 @@
+#include <a8/a8.h>
+
+#include <a8/websocketclient.h>
+
+#ifdef USE_BOOST
+
+static const unsigned char FIN = 0x80;
+static const unsigned char RSV1 = 0x40;
+static const unsigned char RSV2 = 0x20;
+static const unsigned char RSV3 = 0x10;
+static const unsigned char RSV_MASK = RSV1 | RSV2 | RSV3;
+static const unsigned char OPCODE_MASK = 0x0F;
+
+static const unsigned char TEXT_MODE = 0x01;
+static const unsigned char BINARY_MODE = 0x02;
+
+static const unsigned char WEBSOCKET_OPCODE                = 0x0F;
+static const unsigned char WEBSOCKET_FRAME_CONTINUE        = 0x0;
+static const unsigned char WEBSOCKET_FRAME_TEXT            = 0x1;
+static const unsigned char WEBSOCKET_FRAME_BINARY          = 0x2;
+static const unsigned char WEBSOCKET_FRAME_CLOSE           = 0x8;
+static const unsigned char WEBSOCKET_FRAME_PING            = 0x9;
+static const unsigned char WEBSOCKET_FRAME_PONG            = 0xA;
+
+static const unsigned char WEBSOCKET_MASK                  = 0x80;
+static const unsigned char WEBSOCKET_PAYLOAD_LEN           = 0x7F;
+static const unsigned char WEBSOCKET_PAYLOAD_LEN_UINT16    = 126;
+static const unsigned char WEBSOCKET_PAYLOAD_LEN_UINT64    = 127;
+
+static const char* WEB_SOCKET_KEY = "Sec-WebSocket-Key: ";
+static const char* WEB_SOCKET_KEY2 = "Sec-Websocket-Key: ";
+
+static const int DEFAULT_MAX_PACKET_LEN = 1024 * 10;
+static const int DEFAULT_MAX_RECV_BUFFERSIZE = 1024 * 64;
+
+namespace a8
+{
+
+    WebSocketClient::WebSocketClient(std::shared_ptr<asio::io_context> io_context, const std::string& remote_ip, int remote_port)
+    {
+        max_packet_len_ = DEFAULT_MAX_PACKET_LEN;
+        recv_buff_ = (char *)malloc(max_packet_len_ + 1);
+        recv_bufflen_ = 0;
+
+        tcp_client_ = std::make_shared<AsioTcpClient>(io_context, remote_ip, remote_port);
+        decoded_buff_ = (char *)malloc(1024 * 64 + 1);
+        decoded_bufflen_ = 0;
+        tcp_client_->on_error =
+            [this] (a8::AsioTcpClient* socket, int err)
+            {
+                if (on_error) {
+                    on_error(this, err);
+                }
+            };
+        tcp_client_->on_connect =
+            [this] (a8::AsioTcpClient* socket)
+            {
+                std::string data = a8::Format("GET ws://%s:%d/\r\n",
+                                              {socket->GetRemoteAddress(),
+                                                      socket->GetRemotePort()});
+                data += "Upgrade: websocket\r\n";
+                data += "Connection: Upgrade\r\n";
+                data += "Sec-WebSocket-Version: 13\r\n";
+                data += "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==*\r\n";
+                data += "\r\n";
+                socket->SendBuff(data.data(), data.size());
+            };
+        tcp_client_->on_disconnect =
+            [this] (a8::AsioTcpClient* socket)
+            {
+                if (on_disconnect) {
+                    on_disconnect(this);
+                }
+            };
+        tcp_client_->on_socketread =
+            [this] (a8::AsioTcpClient* socket, char* buf, unsigned int buflen)
+            {
+                unsigned int already_read_bytes = 0;
+                do {
+                    if (already_read_bytes < buflen) {
+                        int read_bytes = std::min(buflen - already_read_bytes,
+                                                  (unsigned int)max_packet_len_ - recv_bufflen_);
+                        if (read_bytes > 0) {
+                            memmove(&recv_buff_[recv_bufflen_], buf + already_read_bytes, read_bytes);
+                            recv_bufflen_ += read_bytes;
+                            already_read_bytes += read_bytes;
+                        }
+                    }
+
+                    int offset = 0;
+                    int prev_offset = 0;
+                    do {
+                        prev_offset = offset;
+                        DecodePacket(recv_buff_, offset, recv_bufflen_);
+                    } while (prev_offset < offset && offset < recv_bufflen_);
+
+                    if (offset > 0 && offset < recv_bufflen_){
+                        memmove(recv_buff_, recv_buff_ + offset, recv_bufflen_ - offset);
+                    }
+                    recv_bufflen_ -= offset;
+                    if (recv_bufflen_ >= max_packet_len_) {
+                        //收到超长包
+                        Close();
+                        return;
+                    }
+                } while (already_read_bytes < buflen);
+            };
+    }
+
+    WebSocketClient::~WebSocketClient()
+    {
+        recv_bufflen_ = 0;
+        free(recv_buff_);
+        recv_buff_ = nullptr;
+
+        free(decoded_buff_);
+        decoded_buff_ = nullptr;
+        decoded_bufflen_ = 0;
+    }
+
+    void WebSocketClient::Open()
+    {
+        tcp_client_->Open();
+    }
+
+    void WebSocketClient::Close()
+    {
+        tcp_client_->Close();
+    }
+
+    bool WebSocketClient::IsActive()
+    {
+        return tcp_client_->IsActive();
+    }
+
+    bool WebSocketClient::Connected()
+    {
+        return tcp_client_->Connected();
+    }
+
+    void WebSocketClient::SendBuff(const char* buff, unsigned int bufflen)
+    {
+        if (!handshook_) {
+            abort();
+        }
+        unsigned char szbuff [1024 * 65];
+        szbuff[0] = FIN | BINARY_MODE | 0;
+        int mask_offset = 2;
+        int payloadlen = bufflen;
+        if (payloadlen < 126) {
+            szbuff[1] = payloadlen | 0x80;
+            mask_offset = 2;
+        } else if (payloadlen <= 0xFFFF) {
+            szbuff[1] = 126 | 0x80;
+            szbuff[2] = (payloadlen >> 8) & 0xFF;
+            szbuff[3] = payloadlen & 0xFF;
+            mask_offset = 4;
+        } else {
+            abort();
+        }
+        *((int*)(szbuff + mask_offset)) = rand();
+        for (unsigned i = 0; i < bufflen; ++i) {
+            szbuff[mask_offset + 4 + i] =
+                ((unsigned char)buff[i]) ^ szbuff[mask_offset + (i % 4)] ;
+        }
+        tcp_client_->SendBuff((char*)szbuff, bufflen + mask_offset + 4);
+    }
+
+    void WebSocketClient::ProcessHandShake(char* buf, int& offset, unsigned int buflen)
+    {
+        char* pend = strstr(buf + offset, "\r\n\r\n");
+        if (!pend) {
+            return;
+        }
+        ProcessWsHandShake(buf, offset, buflen);
+    }
+
+    void WebSocketClient::ProcessWsHandShake(char* buf, int& offset, unsigned int buflen)
+    {
+        char* pend = strstr(buf + offset, "\r\n\r\n");
+        if (!pend) {
+            return;
+        }
+        handshook_ = true;
+        offset += pend - buf - offset + strlen("\r\n\r\n");
+        if (on_connect) {
+            on_connect(this);
+        }
+    }
+
+    void WebSocketClient::ProcessUserPacket()
+    {
+        int offset = 0;
+        int prev_offset = 0;
+        do {
+            prev_offset = offset;
+            on_decode_userpacket(decoded_buff_, offset, decoded_bufflen_);
+        } while (prev_offset < offset && offset < decoded_bufflen_);
+
+        if (offset > 0 && offset < decoded_bufflen_){
+            memmove(decoded_buff_, decoded_buff_ + offset, decoded_bufflen_ - offset);
+        }
+        decoded_bufflen_ -= offset;
+        if (decoded_bufflen_ >= max_packet_len_) {
+            //收到超长包
+            Close();
+            return;
+        }
+    }
+
+    void WebSocketClient::DecodeFrame(char* buf, int& offset, unsigned int buflen)
+    {
+        if (offset + 2 > (int)buflen) {
+            return;
+        }
+        char* real_buf = buf + offset;
+        unsigned int ava_len = buflen - offset;
+        unsigned char header = real_buf[0];
+        unsigned char mask_payloadlen = real_buf[1];
+
+        bool is_final_frame = (header & FIN) == FIN;
+        #if 0
+        bool reserved_bits = (header & FIN) == RSV_MASK;
+        #endif
+        unsigned char opcode = header & OPCODE_MASK;
+        #if 0
+        bool opcode_is_control = opcode & 0x8;
+        #endif
+
+        if (opcode == WEBSOCKET_FRAME_CLOSE) {
+            Close();
+            return;
+        }
+        if (opcode != BINARY_MODE) {
+            if (opcode != WEBSOCKET_FRAME_PING) {
+                Close();
+                return;
+            }
+        }
+        if (!is_final_frame) {
+            Close();
+            return;
+        }
+
+        bool is_masked = (mask_payloadlen & 0x80) == 0x80;
+
+        unsigned char payloadlen = mask_payloadlen & 0x7F;
+        unsigned int framelen = 0;
+        int mask_offset = 0;
+
+        if (payloadlen < 126) {
+            framelen = payloadlen;
+            mask_offset = 2;
+        } else if (payloadlen == 126 && ava_len >= 4) {
+            framelen = ntohs( *(u_short*) (real_buf + 2) );
+            mask_offset = 4;
+        } else if (payloadlen == 127 && ava_len >= 8) {
+            //int32 or int64?
+            framelen = ntohl( *(u_long*) (real_buf + 2) );
+            mask_offset = 8;
+        } else {
+            return;
+        }
+        unsigned int real_pkg_len = mask_offset + framelen + (is_masked ? 4 : 0);
+        if (ava_len < real_pkg_len) {
+            return;
+        }
+
+        if (is_masked) {
+            unsigned char *frame_mask = (unsigned char*)(real_buf + mask_offset);
+            memcpy(&decoded_buff_[decoded_bufflen_], real_buf + mask_offset + 4, framelen);
+            for (unsigned int i = 0; i < framelen; i++) {
+                decoded_buff_[decoded_bufflen_ + i] =
+                    (decoded_buff_[ decoded_bufflen_ + i] ^ frame_mask[i%4]);
+            }
+        } else {
+            memcpy(&decoded_buff_[decoded_bufflen_], real_buf + mask_offset, framelen);
+        }
+        decoded_bufflen_ += framelen;
+
+        ProcessUserPacket();
+        offset += real_pkg_len;
+    }
+
+    void WebSocketClient::DecodePacket(char* buf, int& offset, unsigned int buflen)
+    {
+        if (!handshook_) {
+            buf[buflen] = '\0';
+            ProcessHandShake(buf, offset, buflen);
+        } else {
+            DecodeFrame(buf, offset, buflen);
+        }
+    }
+
+}
+
+#endif

a8/websocketclient.h

@@ -0,0 +1,51 @@
+#pragma once
+
+#include <a8/asiotcpclient.h>
+
+#ifdef USE_BOOST
+
+#include <asio.hpp>
+
+using asio::ip::tcp;
+
+namespace a8
+{
+
+    class WebSocketClient
+    {
+    public:
+        WebSocketClient(std::shared_ptr<asio::io_context> io_context, const std::string& remote_ip, int remote_port);
+        virtual ~WebSocketClient();
+
+        std::function<void (a8::WebSocketClient*, int)> on_error;
+        std::function<void (a8::WebSocketClient*)> on_connect;
+        std::function<void (a8::WebSocketClient*)> on_disconnect;
+        std::function<void (char*, int&, unsigned int)> on_decode_userpacket;
+
+        void Open();
+        void Close();
+        bool IsActive();
+        bool Connected();
+        void SendBuff(const char* buff, unsigned int bufflen);
+
+    private:
+        void DecodePacket(char* buf, int& offset, unsigned int buflen);
+
+        void ProcessHandShake(char* buf, int& offset, unsigned int buflen);
+        void ProcessWsHandShake(char* buf, int& offset, unsigned int buflen);
+        void ProcessUserPacket();
+        void DecodeFrame(char* buf, int& offset, unsigned int buflen);
+
+    private:
+        std::shared_ptr<AsioTcpClient> tcp_client_;
+        char *decoded_buff_ = nullptr;
+        int decoded_bufflen_ = 0;
+        bool handshook_ = false;
+        char *recv_buff_ = nullptr;
+        int recv_bufflen_ = 0;
+        int max_packet_len_ = 0;
+    };
+
+}
+
+#endif

a8/websocketsession.cc

@@ -9,31 +9,31 @@
 #include <a8/openssl.h>
 #include <a8/websocketsession.h>
 
-const unsigned char FIN = 0x80;
+static const unsigned char FIN = 0x80;
-const unsigned char RSV1 = 0x40;
+static const unsigned char RSV1 = 0x40;
-const unsigned char RSV2 = 0x20;
+static const unsigned char RSV2 = 0x20;
-const unsigned char RSV3 = 0x10;
+static const unsigned char RSV3 = 0x10;
-const unsigned char RSV_MASK = RSV1 | RSV2 | RSV3;
+static const unsigned char RSV_MASK = RSV1 | RSV2 | RSV3;
-const unsigned char OPCODE_MASK = 0x0F;
+static const unsigned char OPCODE_MASK = 0x0F;
 
-const unsigned char TEXT_MODE = 0x01;
+static const unsigned char TEXT_MODE = 0x01;
-const unsigned char BINARY_MODE = 0x02;
+static const unsigned char BINARY_MODE = 0x02;
 
-const unsigned char WEBSOCKET_OPCODE                = 0x0F;
+static const unsigned char WEBSOCKET_OPCODE                = 0x0F;
-const unsigned char WEBSOCKET_FRAME_CONTINUE        = 0x0;
+static const unsigned char WEBSOCKET_FRAME_CONTINUE        = 0x0;
-const unsigned char WEBSOCKET_FRAME_TEXT            = 0x1;
+static const unsigned char WEBSOCKET_FRAME_TEXT            = 0x1;
-const unsigned char WEBSOCKET_FRAME_BINARY          = 0x2;
+static const unsigned char WEBSOCKET_FRAME_BINARY          = 0x2;
-const unsigned char WEBSOCKET_FRAME_CLOSE           = 0x8;
+static const unsigned char WEBSOCKET_FRAME_CLOSE           = 0x8;
-const unsigned char WEBSOCKET_FRAME_PING            = 0x9;
+static const unsigned char WEBSOCKET_FRAME_PING            = 0x9;
-const unsigned char WEBSOCKET_FRAME_PONG            = 0xA;
+static const unsigned char WEBSOCKET_FRAME_PONG            = 0xA;
 
-const unsigned char WEBSOCKET_MASK                  = 0x80;
+static const unsigned char WEBSOCKET_MASK                  = 0x80;
-const unsigned char WEBSOCKET_PAYLOAD_LEN           = 0x7F;
+static const unsigned char WEBSOCKET_PAYLOAD_LEN           = 0x7F;
-const unsigned char WEBSOCKET_PAYLOAD_LEN_UINT16    = 126;
+static const unsigned char WEBSOCKET_PAYLOAD_LEN_UINT16    = 126;
-const unsigned char WEBSOCKET_PAYLOAD_LEN_UINT64    = 127;
+static const unsigned char WEBSOCKET_PAYLOAD_LEN_UINT64    = 127;
 
-const char* WEB_SOCKET_KEY = "Sec-WebSocket-Key: ";
+static const char* WEB_SOCKET_KEY = "Sec-WebSocket-Key: ";
-const char* WEB_SOCKET_KEY2 = "Sec-Websocket-Key: ";
+static const char* WEB_SOCKET_KEY2 = "Sec-Websocket-Key: ";
 
 namespace a8
 {
@@ -188,6 +188,11 @@ namespace a8
         }
     }
 
+    /*
+      finbit|opcode|falgs
+      unsigned char header FIN|BINARY_MODE
+      unsigned char mask_payloadlen
+     */
     void WebSocketSession::DecodeFrame(char* buf, int& offset, unsigned int buflen)
     {
         if (offset + 2 > (int)buflen) {