使用websocketpp進行websocket通信

websocketpp介紹

websocketpp是一個只有頭文件的支持websocket協議的C++開源庫，支持websocket客戶端和服務器功能，網絡傳輸模塊基於boost::asio

提供 server 功能的 websocketpp::server 和提供 client 功能的 websocketpp:client 都繼承自基類 websocketpp::endpoint , endpoint提供了一些通用的功能函數：

void set_access_channels(log::level channels);//設置日志級別
void clear_access_channels(log::level channels)//屏蔽某個級別的日志

void set_open_handler(open_handler h)；//設置打開連接時的回調函數
void set_close_handler(close_handler h)；//設置關閉連接時的回調函數
void set_fail_handler(fail_handler h)；//設置連接失敗時的回調函數
void set_message_handler(message_handler h)；//設置收到消息時的回調函數

服務器代碼

#include <websocketpp/config/asio_no_tls.hpp>
#include <websocketpp/server.hpp>

#include <functional>

typedef websocketpp::server<websocketpp::config::asio> server;

class utility_server {
public:
    utility_server() {
         // 設置log
        m_endpoint.set_error_channels(websocketpp::log::elevel::all);
        m_endpoint.set_access_channels(websocketpp::log::alevel::all ^ websocketpp::log::alevel::frame_payload);

        // 初始化Asio
        m_endpoint.init_asio();

        // 設置消息回調為echo_handler
        m_endpoint.set_message_handler(std::bind(
            &utility_server::echo_handler, this,
            std::placeholders::_1, std::placeholders::_2
        ));
    }

    void echo_handler(websocketpp::connection_hdl hdl, server::message_ptr msg) {
        // 發送消息
        m_endpoint.send(hdl, msg->get_payload(), msg->get_opcode());
    }

    void run() {
        // 監聽端口 9002
        m_endpoint.listen(9002);

       
        m_endpoint.start_accept();

        // 開始Asio事件循環
        m_endpoint.run();
    }
private:
    server m_endpoint;
};

int main() {
    utility_server s;
    s.run();
    return 0;
}

運行后執行到 s.run() 當前線程會進入Asio的消息循環，可以調用s.top()退出

客戶端代碼

#include <websocketpp/config/asio_no_tls_client.hpp>
#include <websocketpp/client.hpp>
#include <iostream>

typedef websocketpp::client<websocketpp::config::asio_client> client;

using websocketpp::lib::placeholders::_1;
using websocketpp::lib::placeholders::_2;
using websocketpp::lib::bind;

typedef websocketpp::config::asio_client::message_type::ptr message_ptr;

void on_open(client *c, websocketpp::connection_hdl hdl)
{
    std::string msg = "hello";
    c->send(hdl, msg, websocketpp::frame::opcode::text);
    c->get_alog().write(websocketpp::log::alevel::app, "Tx: " + msg);
}

void on_message(client *c, websocketpp::connection_hdl hdl, message_ptr msg)
{
    std::cout << "on_message called with hdl: " << hdl.lock().get()
              << " and message: " << msg->get_payload()
              << std::endl;
    websocketpp::lib::error_code ec;
    //c->send(hdl,msg->get_payload(),msg->get_opcode(),ec);
    if(ec)
    {
        std::cout << "Echo failed because " << ec.message() << std::endl;
    }
}

//定時器回調函數
void Timeout(client *c, websocketpp::connection_hdl &hdl, boost::asio::deadline_timer *pt, const boost::system::error_code &ec)
{
    if(ec)
    {
        std::cout << "timer is cancel " << std::endl;
        return;
    }
    static int count = 0;
    c->send(hdl, "hello", websocketpp::frame::opcode::text);
    count++;
    if(count > 5)//定時器觸發五次后關閉連接
    {
        c->close(hdl, websocketpp::close::status::normal, "");
        return;
    }
    pt->expires_at(pt->expires_at() + boost::posix_time::seconds(5));
    pt->async_wait(bind(Timeout, c, hdl, pt, ::_1));

}

int main(int argc, char *argv[])
{
    client c;

    std::string uri = "ws://xx.xx.xx.xx:xxx";

    c.set_access_channels(websocketpp::log::alevel::all);
    c.clear_access_channels(websocketpp::log::alevel::frame_payload);
    c.clear_access_channels(websocketpp::log::alevel::frame_header);

    // 初始化 ASIO
    c.init_asio();

    // 注冊消息回調

    c.set_message_handler(bind(&on_message, &c, ::_1, ::_2));
    c.set_open_handler(bind(&on_open, &c, _1));

    websocketpp::lib::error_code ec;
    client::connection_ptr con = c.get_connection(uri, ec);
    con->add_subprotocol("janus-protocol");
    if(ec)
    {
        std::cout << "could not create connection because: " << ec.message() << std::endl;
        return 0;
    }

    auto hdl = con->get_handle();
    c.connect(con);
    
    boost::asio::deadline_timer t(c.get_io_service(), boost::posix_time::seconds(5)); //設置一個5s超時的定時器
    t.async_wait(bind(&Timeout, &c, hdl, &t, ::_1));

    std::thread th([&c] { c.run(); });
   
    //休眠13s后取消定時器並關閉連接
    sleep(13);
    t.cancel();
    c.close(hdl, websocketpp::close::status::normal, "");
    
    th.join();
}

我們可以利用 Asio 的一些其它組件，如定時器等；初始化定時器傳入 c.get_io_service() 在一個io循環中處理定時事件和其它事件
當客戶端調用 close 關閉連接時，則自動退出c.run()開啟的循環

重復關閉和打開連接

#include <websocketpp/config/asio_no_tls_client.hpp>
#include <websocketpp/client.hpp>
#include <iostream>

typedef websocketpp::client<websocketpp::config::asio_client> client;

using websocketpp::lib::placeholders::_1;
using websocketpp::lib::placeholders::_2;

typedef websocketpp::config::asio_client::message_type::ptr message_ptr;
class connection
{
public:

  void on_open(websocketpp::connection_hdl hdl)
  {
      std::string msg = "hello";
      c.send(hdl, msg, websocketpp::frame::opcode::text);
      c.get_alog().write(websocketpp::log::alevel::app, "Tx: " + msg);
  }

  void on_message(websocketpp::connection_hdl hdl, message_ptr msg)
  {
      std::cout << "on_message called with hdl: " << hdl.lock().get()
                << " and message: " << msg->get_payload()
                << std::endl;
      websocketpp::lib::error_code ec;
      if(ec)
      {
          std::cout << "Echo failed because " << ec.message() << std::endl;
      }
  }

  int init()
  {
      uri = "ws://xx.xx.xx.xx:xx";

      c.set_access_channels(websocketpp::log::alevel::all);
      c.clear_access_channels(websocketpp::log::alevel::frame_payload);
      c.clear_access_channels(websocketpp::log::alevel::frame_header);

      c.init_asio();


      c.set_message_handler(websocketpp::lib::bind(&connection::on_message, this,::_1, ::_2));
      c.set_open_handler(websocketpp::lib::bind(&connection::on_open, this, _1));

      c.start_perpetual();
      thread_=websocketpp::lib::make_shared<websocketpp::lib::thread>(&client::run,&c);

  }

  void connect()
  {
      websocketpp::lib::error_code ec;
      client::connection_ptr con = c.get_connection(uri, ec);
      if(ec)
      {
          std::cout << "could not create connection because: " << ec.message() << std::endl;
          return ;
      }

      hdl_ = con->get_handle();
      c.connect(con);

  }
  void close()
  {
      c.close(hdl_, websocketpp::close::status::normal, "");
  }

  void terminate()
  {
      c.stop_perpetual();
      thread_->join();
  }
private:
  client c;
  websocketpp::lib::shared_ptr<websocketpp::lib::thread> thread_;
  websocketpp::connection_hdl hdl_;
  std::string uri;

};

需要重復打開關閉連接時，只調用一次init_asio()函數，然后調用start_perpetual()將endpoint設置為永久的，不會在連接斷開時自動退出。需要結束循環，調用stop_perpetual()