broker

#include <websocketpp/config/asio.hpp>
#include <websocketpp/server.hpp>
#include <iostream>
#include <websocketpp/common/thread.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/classification.hpp>

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

using websocketpp::connection_hdl;

using websocketpp::lib::thread;
using websocketpp::lib::mutex;
using websocketpp::lib::unique_lock;
using websocketpp::lib::condition_variable;

enum action_type {
    SUBSCRIBE,
    UNSUBSCRIBE,
    MESSAGE
};

struct action {
    action(action_type t, connection_hdl h) : type(t), hdl(h) {}
    action(action_type t, server::message_ptr m) : type(t), msg(m) {}

    action_type type;
    websocketpp::connection_hdl hdl;
    server::message_ptr msg;
};

void parse_key_value_string(std::map<std::string,std::string> out_map, std::string in_list) {
    std::vector<std::string> kvs;
    boost::split(kvs,in_list,boost::is_any_of("\n"));
    for(auto kv: kvs) {
        std::vector<std::string> parts;
        boost::split(parts,kv,boost::is_any_of(":"));
        if (parts.size()==2) {
            out_map[parts[0]] = parts[1];
        }
    }
}

class broadcast_server {
public:
    broadcast_server() {
        // Initialize Asio Transport
        m_server.init_asio();

        // Register handler callbacks
        m_server.set_open_handler(bind(&broadcast_server::on_open,this,::_1));
        m_server.set_close_handler(bind(&broadcast_server::on_close,this,::_1));
        m_server.set_message_handler(bind(&broadcast_server::on_message,this,::_1,::_2));
    }

    void run(uint16_t port) {
        // listen on specified port
        m_server.listen(port);

        // Start the server accept loop
      m_server.start_accept();

	    // Start the ASIO io_service run loop
        try {
            m_server.run();
        } catch (const std::exception & e) {
            std::cout << e.what() << std::endl;
        } catch (websocketpp::lib::error_code e) {
            std::cout << e.message() << std::endl;
        } catch (...) {
            std::cout << "other exception" << std::endl;
        }
    }

    void on_open(connection_hdl hdl) {
        unique_lock<mutex> lock(m_action_lock);
        //std::cout << "on_open" << std::endl;
        m_actions.push(action(SUBSCRIBE,hdl));
        lock.unlock();
        m_action_cond.notify_one();
    }

    void on_close(connection_hdl hdl) {
        unique_lock<mutex> lock(m_action_lock);
        //std::cout << "on_close" << std::endl;
        m_actions.push(action(UNSUBSCRIBE,hdl));
        lock.unlock();
        m_action_cond.notify_one();
    }

    void on_message(connection_hdl hdl, server::message_ptr msg) {
        // queue message up for sending by processing thread
        unique_lock<mutex> lock(m_action_lock);
        //std::cout << "on_message" << std::endl;
        m_actions.push(action(MESSAGE,msg));
        lock.unlock();
        m_action_cond.notify_one();
    }

    void process_messages() {
        while(1) {
            unique_lock<mutex> lock(m_action_lock);

            while(m_actions.empty()) {
                m_action_cond.wait(lock);
            }

            action a = m_actions.front();
            m_actions.pop();

            lock.unlock();

            if (a.type == SUBSCRIBE) {
                unique_lock<mutex> lock(m_connection_lock);
                m_connections.insert(a.hdl);
            } else if (a.type == UNSUBSCRIBE) {
                unique_lock<mutex> lock(m_connection_lock);
                m_connections.erase(a.hdl);
            } else if (a.type == MESSAGE) {
                unique_lock<mutex> lock(m_connection_lock);
                
                std::map<std::string,std::string> cmds;
                parse_key_value_string(cmds,a.msg->get_payload());
                auto cmd_key = cmds.find("CMD");
                if (cmd_key!=cmds.end()) {
                    std::cout << cmd_key->second;
                }

                con_list::iterator it;
                for (it = m_connections.begin(); it != m_connections.end(); ++it) {
                   
                    m_server.send(*it,a.msg);
                }
            } else {
                // undefined.
            }
        }
    }
private:
    typedef std::set<connection_hdl,std::owner_less<connection_hdl>> con_list;
    typedef std::map<connection_hdl,std::string,std::owner_less<connection_hdl>> conn_to_user_map_t;
    typedef std::map<std::string,connection_hdl> user_to_conn_map_t;

    server m_server;
    con_list m_connections;
    std::queue<action> m_actions;
    
    conn_to_user_map_t conn_to_user_map;
    user_to_conn_map_t user_to_conn_map;

    mutex m_action_lock;
    mutex m_connection_lock;
    condition_variable m_action_cond;
};

int main() {
	try {
	broadcast_server server;

	// Start a thread to run the processing loop
	//thread t(bind(&broadcast_server::process_messages,&server));
	thread t([&server]{
	    server.process_messages();
	});

	// Run the asio loop with the main thread
	server.run(80);

	t.join();

	} catch (std::exception & e) {
	    std::cout << e.what() << std::endl;
	}
}