Advertisement
Guest User

Untitled

a guest
Mar 30th, 2018
275
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
C++ 5.17 KB | None | 0 0
  1. #include <websocketpp/config/asio_no_tls.hpp>
  2.  
  3. #include <websocketpp/server.hpp>
  4.  
  5. #include <iostream>
  6. #include <set>
  7. #include <map>
  8.  
  9. /*#include <boost/thread.hpp>
  10. #include <boost/thread/mutex.hpp>
  11. #include <boost/thread/condition_variable.hpp>*/
  12. #include <websocketpp/common/thread.hpp>
  13.  
  14. typedef websocketpp::server<websocketpp::config::asio> server;
  15.  
  16. using websocketpp::connection_hdl;
  17. using websocketpp::lib::placeholders::_1;
  18. using websocketpp::lib::placeholders::_2;
  19. using websocketpp::lib::bind;
  20.  
  21. using websocketpp::lib::thread;
  22. using websocketpp::lib::mutex;
  23. using websocketpp::lib::lock_guard;
  24. using websocketpp::lib::unique_lock;
  25. using websocketpp::lib::condition_variable;
  26.  
  27. /* on_open insert connection_hdl into channel
  28.  * on_close remove connection_hdl from channel
  29.  * on_message queue send to all channels
  30.  */
  31.  
  32. enum action_type {
  33.     SUBSCRIBE,
  34.     UNSUBSCRIBE,
  35.     MESSAGE
  36. };
  37.  
  38. struct action {
  39.     action(action_type t, connection_hdl h) : type(t), hdl(h) {}
  40.     action(action_type t, connection_hdl h, server::message_ptr m)
  41.       : type(t), hdl(h), msg(m) {}
  42.  
  43.     action_type type;
  44.     websocketpp::connection_hdl hdl;
  45.     server::message_ptr msg;
  46. };
  47.  
  48. class broadcast_server {
  49. public:
  50.     broadcast_server() {
  51.         // Initialize Asio Transport
  52.         m_server.init_asio();
  53.  
  54.         // Register handler callbacks
  55.         m_server.set_open_handler(bind(&broadcast_server::on_open,this,::_1));
  56.         m_server.set_close_handler(bind(&broadcast_server::on_close,this,::_1));
  57.         m_server.set_message_handler(bind(&broadcast_server::on_message,this,::_1,::_2));
  58.     }
  59.  
  60.     void run(uint16_t port) {
  61.         // listen on specified port
  62.         m_server.listen(port);
  63.  
  64.         // Start the server accept loop
  65.         m_server.start_accept();
  66.  
  67.         // Start the ASIO io_service run loop
  68.         try {
  69.             m_server.run();
  70.         } catch (const std::exception & e) {
  71.             std::cout << e.what() << std::endl;
  72.         }
  73.     }
  74.  
  75.     void on_open(connection_hdl hdl) {
  76.         {
  77.             lock_guard<mutex> guard(m_action_lock);
  78.             //std::cout << "on_open" << std::endl;
  79.             m_actions.push(action(SUBSCRIBE,hdl));
  80.         }
  81.         m_action_cond.notify_one();
  82.     }
  83.  
  84.     void on_close(connection_hdl hdl) {
  85.         {
  86.             lock_guard<mutex> guard(m_action_lock);
  87.             //std::cout << "on_close" << std::endl;
  88.             m_actions.push(action(UNSUBSCRIBE,hdl));
  89.         }
  90.         m_action_cond.notify_one();
  91.     }
  92.  
  93.     void on_message(connection_hdl hdl, server::message_ptr msg) {
  94.         // queue message up for sending by processing thread
  95.         {
  96.             lock_guard<mutex> guard(m_action_lock);
  97.             //std::cout << "on_message" << std::endl;
  98.             m_actions.push(action(MESSAGE,hdl,msg));
  99.         }
  100.         m_action_cond.notify_one();
  101.     }
  102.  
  103.     void process_messages() {
  104.         while(1) {
  105.             unique_lock<mutex> lock(m_action_lock);
  106.  
  107.             while(m_actions.empty()) {
  108.                 m_action_cond.wait(lock);
  109.             }
  110.  
  111.             action a = m_actions.front();
  112.             m_actions.pop();
  113.  
  114.             lock.unlock();
  115.  
  116.             if (a.type == SUBSCRIBE) {
  117.                 lock_guard<mutex> guard(m_connection_lock);
  118.                 //m_connections.insert(a.hdl);
  119.                 if (m_NumOfClient > 2) { // 若房間大於2人
  120.                     m_iKey = m_iKey + 1; // 開新房
  121.                     m_NumOfClient = 0;   // clean
  122.                 }
  123.                 m_HubTo_list[m_iKey].insert(a.hdl);
  124.                
  125.             } else if (a.type == UNSUBSCRIBE) {
  126.                 lock_guard<mutex> guard(m_connection_lock);
  127.                 m_connections.erase(a.hdl);
  128.             } else if (a.type == MESSAGE) {
  129.                 lock_guard<mutex> guard(m_connection_lock);
  130.  
  131.                 //con_list::iterator it;
  132.                 for (HubToCon_list::iterator it = m_HubTo_list.begin(); it != m_HubTo_list.end(); it++) {
  133.                     con_list set = it->second;
  134.                     for (con_list::iterator getit = set.begin(); getit != set.end(); ++getit) {
  135.                         m_server.send(*getit, a.msg);
  136.                     }
  137.                 }
  138.                
  139.               /*  for (it = m_connections.begin(); it != m_connections.end(); ++it) {
  140.                     m_server.send(*it,a.msg);
  141.                 }*/
  142.             } else {
  143.                 // undefined.
  144.             }
  145.         }
  146.     }
  147. private:
  148.     typedef std::set<connection_hdl,std::owner_less<connection_hdl> > con_list;
  149.     typedef std::map<int , con_list> HubToCon_list; // int == NameOfChatroom , con_list == list of client in this room
  150.  
  151.     server m_server;
  152.     con_list m_connections;
  153.     //
  154.     HubToCon_list m_HubTo_list ; // 建立一個多通道(房) 聊天
  155.     int m_iKey = 0;
  156.     int m_NumOfClient = 0;
  157.     //
  158.     std::queue<action> m_actions;
  159.  
  160.     mutex m_action_lock;
  161.     mutex m_connection_lock;
  162.     condition_variable m_action_cond;
  163. };
  164.  
  165. int main() {
  166.     try {
  167.     broadcast_server server_instance;
  168.  
  169.     // Start a thread to run the processing loop
  170.     thread t(bind(&broadcast_server::process_messages,&server_instance));
  171.  
  172.     // Run the asio loop with the main thread
  173.     server_instance.run(9002);
  174.  
  175.     t.join();
  176.  
  177.     } catch (websocketpp::exception const & e) {
  178.         std::cout << e.what() << std::endl;
  179.     }
  180. }
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement