WebSocketClient.h

Go to the documentation of this file.

 
#ifndef SUPPORTLIB_WEBSOCKETCLIENT_H
#define SUPPORTLIB_WEBSOCKETCLIENT_H
#include "Observer.h"
#include "Exception.h"
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/websocket/ssl.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <cstdlib>
#include <iostream>
#include <string>
#include <thread>
 
namespace giri {
    using tcp = boost::asio::ip::tcp;
    namespace ssl = boost::asio::ssl;
    namespace websocket = boost::beast::websocket;
 
    class WebSocketClientException : public ExceptionBase
    {
    public:
        WebSocketClientException(const std::string &msg) : ExceptionBase(msg) {}; 
        using SPtr = std::shared_ptr<WebSocketClientException>;
        using UPtr = std::unique_ptr<WebSocketClientException>;
        using WPtr = std::weak_ptr<WebSocketClientException>;
    };
 
    class WebSocketClient : public Observable<WebSocketClient>
    {
    public:
        WebSocketClient(const std::string& host, const std::string& port, bool ssl = false, const size_t numThreads = 1, const std::string& resource = "/") : 
            m_Host(host), 
            m_Port(port), 
            m_Resource(resource),
            m_SSL(ssl),
            m_Ioc(numThreads)
        {
            auto const results = m_Resolver.resolve(m_Host, m_Port);
            if(m_SSL)
            {
                boost::asio::connect(m_Wss.next_layer().next_layer(), results.begin(), results.end());
                m_Wss.next_layer().handshake(ssl::stream_base::client);
                m_Wss.handshake(host, m_Resource);
                m_Wss.text(true);
            }
            else 
            {
                boost::asio::connect(m_Ws.next_layer(), results.begin(), results.end());
                m_Ws.handshake(host, m_Resource);
                m_Ws.text(true);
            }
            m_Threads.reserve(numThreads);
            for(auto i = numThreads; i > 0; --i)
                m_Threads.emplace_back(std::thread([this]{ m_Ioc.run();}));
        }
        void send(const std::string& msg){
            if(!m_Ec)
                if(m_SSL) 
                    {if(m_Wss.is_open()){m_Wss.write(boost::asio::buffer(msg));}}
                else
                    {if(m_Ws.is_open()){m_Ws.write(boost::asio::buffer(msg));}}
        }
        std::string getMessage(){ 
            return m_Message;
        }
        std::string getHost(){ 
            return m_Host;
        }
        std::string getPort(){ 
            return m_Port;
        }
        bool getSSL(){ 
            return m_SSL;
        }
        boost::system::error_code getError(){
            return m_Ec;
        }
        void run(){
            if(m_SSL)
                {if(m_Wss.is_open()){m_Wss.async_read(m_Buffer, std::bind(&WebSocketClient::on_read, this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));}}
            else
                {if(m_Ws.is_open()){m_Ws.async_read(m_Buffer, std::bind(&WebSocketClient::on_read, this->shared_from_this(), std::placeholders::_1, std::placeholders::_2));}}
        }
        std::string receive(){        
            if(!m_Ec)
                if(m_SSL) 
                    {if(m_Wss.is_open()){m_Wss.read(m_Buffer);}}
                else
                    {if(m_Ws.is_open()){m_Ws.read(m_Buffer);}}
            m_Message = boost::beast::buffers_to_string(m_Buffer.data());
            return m_Message;
        }
        void close() {
            if(!m_Ec)
                if(m_SSL) 
                    {if(m_Wss.is_open()){m_Wss.close(websocket::close_code::normal);}}
                else
                    {if(m_Ws.is_open()){m_Ws.close(websocket::close_code::normal);}}
        }
        using SPtr = std::shared_ptr<WebSocketClient>;
        using UPtr = std::unique_ptr<WebSocketClient>;
    private:
        void on_read(boost::system::error_code ec, std::size_t bytes_transferred) {
            m_Ec = ec;
            m_Message.clear();
            if(!ec){
                m_Message = boost::beast::buffers_to_string(m_Buffer.data());
            }
            m_Buffer.consume(m_Buffer.size()); // clear buffer
            notify(); // notify all subscribed observers
            if(!ec)
                run(); // wait for new message
        }
        std::string m_Host;
        std::string m_Port;
        std::string m_Resource;
        std::string m_Message;
        bool m_SSL;
        boost::asio::io_context m_Ioc;
        tcp::resolver m_Resolver{m_Ioc};
        ssl::context m_Ctx{ssl::context::sslv23_client};
        websocket::stream<tcp::socket> m_Ws{m_Ioc};
        websocket::stream<ssl::stream<tcp::socket>> m_Wss{m_Ioc, m_Ctx};
        boost::beast::multi_buffer m_Buffer;
        boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_WG = boost::asio::make_work_guard(m_Ioc);
        std::vector<std::thread> m_Threads;
        boost::system::error_code m_Ec;
    };
}
#endif //SUPPORTLIB_WEBSOCKETCLIENT_H