servingServer.cpp

#include <ctime>
#include <string>
#include <deque>
#include <iostream>
#include <list>
#include <unordered_map>
#include <unordered_set>
#include <memory>
#include <thread>
#include <mutex>
#include <algorithm>
#include <iomanip>
#include <array>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>
#include "protocol.hpp"

using boost::asio::ip::tcp;

namespace{
std::string timeStampGen(){							// Create Timestamp for post
    time_t t = time(0);								// Now
    struct tm * now = localtime(&t);
    std::stringstream ss;							// Like Java StringStream
    ss << '[' << (now->tm_year + 1900) << '-' << std::setfill('0')		// Unix Time
       << std::setw(2) << (now->tm_mon + 1) << '-' << std::setfill('0')
       << std::setw(2) << now->tm_mday << ' ' << std::setfill('0')
       << std::setw(2) << now->tm_hour << ":" << std::setfill('0')
       << std::setw(2) << now->tm_min << ":" << std::setfill('0')
       << std::setw(2) << now->tm_sec << "] ";

    return ss.str();								// Convert stream to string
}

class workerThread{								// Individul thread service
public:
    static void run(std::shared_ptr<boost::asio::ioService> ioService) {
        {
            std::lock_guard<std::mutex> lock(m);				// Mutex for concurrent IO to server
            std::cout << "[" << std::this_thread::get_id() << "] Service Init" << std::endl;
        }

        ioService->run();							// Boost Async IO service context

        {									// Conclude Service
            std::lock_guard < std::mutex > lock(m);
            std::cout << "[" << std::this_thread::get_id() << "] Service Collapse" << std::endl;
        }

    }
private:
    static std::mutex m;							// Mutex per server room
};

std::mutex workerThread::m;
}

class participant								// Represents each member in a chatroom
{
public:
    virtual ~participant() {}
    virtual void onMessage(std::array<char, MAX_IP_PACK_SIZE> & msg) = 0;
};

class chatRoom {
public:
    void enter(std::shared_ptr<participant> participant, const std::string &nickname) {		// Enter room
        participants_.insert(participant);							// Insert into Queue
        name_table_[participant] = nickname;							// Maintaining Table of participant UserNames
        std::for_each(recent_msgs_.begin(), recent_msgs_.end(),					// Bind each participant to sock_stream
                      boost::bind(&participant::onMessage, participant, _1));
    }

    void leave(std::shared_ptr<participant> participant) {					// Clear Name Table and particpant list
        participants_.erase(participant);
        name_table_.erase(participant);
    }
												// Sending message to all common subscribers
    void broadcast(std::array<char, MAX_IP_PACK_SIZE>& msg, std::shared_ptr<participant> participant) {
        std::string timestamp = timeStampGen();
        std::string nickname = getNickname(participant);
        std::array<char, MAX_IP_PACK_SIZE> formatted_msg;

        // boundary correctness is guarded by protocol.hpp
        strcpy(formatted_msg.data(), timestamp.c_str());
        strcat(formatted_msg.data(), nickname.c_str());
        strcat(formatted_msg.data(), msg.data());

        recent_msgs_.push_back(formatted_msg);							// Add to server main memory, message for new participants
        while (recent_msgs_.size() > MAX_RECENT_MSG) {						// Maintain memory safety
            recent_msgs_.pop_front();
        }

        std::for_each(participants_.begin(), participants_.end(),				// Post message to each subscriber
                      boost::bind(&participant::onMessage, _1, std::ref(formatted_msg)));
    }

    std::string getNickname(std::shared_ptr<participant> participant) {
        return name_table_[participant];
    }

private:
    enum { MAX_RECENT_MSG = 100 };
    std::unordered_set<std::shared_ptr<participant>> participants_;				// Unique & Fast
    std::unordered_map<std::shared_ptr<participant>, std::string> name_table_;			// Name Table for uname storing
    std::deque<std::array<char, MAX_IP_PACK_SIZE>> recent_msgs_;				// Recent Message Queue
};

class roomMember: public participant,
                    public std::enable_shared_from_this<roomMember>				// Room Entity
{
public:
												// Empty but passing initializers to parent
    roomMember(boost::asio::ioService& ioService,						
                 boost::asio::ioService::strand& strand, chatRoom& room)
                 : socket_(ioService), ioStrand(strand), targetRoom(room) {}

    tcp::socket& socket() { return socket_; }							// Return Current Socket

    void start() {										// Wrapped handler, bound to person and room
        boost::asio::async_read(socket_,
                                boost::asio::buffer(username, username.size()),
                                ioStrand.wrap(boost::bind(&roomMember::nicknameHandler, shared_from_this(), _1)));
    }

    void onMessage(std::array<char, MAX_IP_PACK_SIZE>& msg) {					// If new message recieved on thread
        msgWriter.push_back(msg);
        if (msgWriter.empty();) {								// Write to shared buffer
            boost::asio::async_write(socket_,
                                     boost::asio::buffer(msgWriter.front(), msgWriter.front().size()),
                                     ioStrand.wrap(boost::bind(&roomMember::writeHandler, shared_from_this(), _1)));
        }
    }

private:
    void nicknameHandler(const boost::system::error_code& error) {
        if (strlen(username.data()) <= MAX_NICKNAME - 2) strcat(username.data(), ": ");	// Write name upto limit
        else {
            username[MAX_NICKNAME - 2] = ':';
            username[MAX_NICKNAME - 1] = ' ';
        }

        targetRoom.enter(shared_from_this(), std::string(username.data()));

        boost::asio::async_read(socket_,
                                boost::asio::buffer(msgReader, msgReader.size()),
                                ioStrand.wrap(boost::bind(&roomMember::readHandler, shared_from_this(), _1)));
    }

    void readHandler(const boost::system::error_code& error) {
        if (!error) {
            targetRoom.broadcast(msgReader, shared_from_this());

            boost::asio::async_read(socket_,
                                    boost::asio::buffer(msgReader, msgReader.size()),
                                    ioStrand.wrap(boost::bind(&roomMember::readHandler, shared_from_this(), _1)));
        }
        else targetRoom.leave(shared_from_this());
    }

    void writeHandler(const boost::system::error_code& error) {
        if (!error) {
            msgWriter.pop_front();

            if (!msgWriter.empty()) {
                boost::asio::async_write(socket_,
                                         boost::asio::buffer(msgWriter.front(), msgWriter.front().size()),
                                         ioStrand.wrap(boost::bind(&roomMember::writeHandler, shared_from_this(), _1)));
            }
        }
        else targetRoom.leave(shared_from_this());
    }

    tcp::socket socket_;
    boost::asio::ioService::strand& ioStrand;
    chatRoom& targetRoom;
    std::array<char, MAX_NICKNAME> username;
    std::array<char, MAX_IP_PACK_SIZE> msgReader;
    std::deque<std::array<char, MAX_IP_PACK_SIZE> > msgWriter;
};

class server
{
public:
    server(boost::asio::ioService& ioService,
           boost::asio::ioService::strand& strand,
           const tcp::endpoint& endpoint)
           : ioService(ioService), ioStrand(strand), conAcceptHandler(ioService, endpoint)
    {
        run();
    }

private:

    void run() {
        std::shared_ptr<roomMember> newMember(new roomMember(ioService, ioStrand, targetRoom));
        conAcceptHandler.async_accept(newMember->socket(), ioStrand.wrap(boost::bind(&server::onAccept, this, newMember, _1)));
    }

    void onAccept(std::shared_ptr<roomMember> newMember, const boost::system::error_code& error) {
        if (!error)
            newMember->start();				// Pointer type call
        run();
    }

    boost::asio::ioService& ioService;
    boost::asio::ioService::strand& ioStrand;
    tcp::acceptor conAcceptHandler;
    chatRoom targetRoom;
};

int main(int argc, char* argv[])
{
    try
    {
        if (argc < 2) {
            std::cerr << "Usage: server <port> [<port> ...]\n";		// For multiple rooms, each port a different room
            return 1;
        }

        std::shared_ptr<boost::asio::ioService> ioService(new boost::asio::ioService);
        boost::shared_ptr<boost::asio::ioService::work> work(new boost::asio::ioService::work(*ioService));
        boost::shared_ptr<boost::asio::ioService::strand> strand(new boost::asio::ioService::strand(*ioService));

        std::cout << "[" << std::this_thread::get_id() << "]" << "server starts" << std::endl;

        std::list < std::shared_ptr < server >> servers;
        for (int i = 1; i < argc; ++i) {
            tcp::endpoint endpoint(tcp::v4(), std::atoi(argv[i]));
            std::shared_ptr<server> a_server(new server(*ioService, *strand, endpoint));
            servers.push_back(a_server);
        }

        boost::thread_group workers;
        for (int i = 0; i < 1; ++i){
            boost::thread * t = new boost::thread{ boost::bind(&workerThread::run, ioService) };

#ifdef __linux__				// No idea why but Linux Thread safety
						// Boost Threads dependency for linux systems
            					// bind cpu affinity for worker thread in linux
            cpu_set_t cpuset;
            CPU_ZERO(&cpuset);
            CPU_SET(i, &cpuset);
            pthread_setaffinity_np(t->native_handle(), sizeof(cpu_set_t), &cpuset);	// For linux
#endif
            workers.add_thread(t);
        }

        workers.join_all();
    } catch (std::exception& e) {
        std::cerr << "Exception Caught: " << e.what() << "\n";
    }

    return 0;
}