sender.cpp

Go to the documentation of this file.

/*
 * sender.cpp
 *
 *  Created on: Jan 10, 2012
 *      Author: sushil
 */

#include "DEBUG.h"
#include <messages.h>
#include <sender.h>
#include <socket.h>

#include <boost/asio.hpp>
#include <boost/thread.hpp>

#include<iostream>

using namespace FastEcslent;
using namespace boost::asio::ip;

void FastEcslent::Sender::init() { //initialize sockets, public
        DEBUG(std::cout << "Initializing Sender before starting sender thread. port: " << port << std::endl;)
        quit = false;
        sleepTime = new boost::posix_time::milliseconds(500);

}


void FastEcslent::Sender::run() { //start thread, public
        senderThread = boost::thread(&FastEcslent::Sender::runThread, this);
        DEBUG(std::cout << "Started Sender thread's run method" << senderThread.get_id() << std::endl;)
}

void FastEcslent::Sender::addMessage(Message *m) {
        DEBUG(std::cout << "Sender Added message of Type: " << (int) m->head.msgType << std::endl;)
        sendBuffer.push_back(m);
}

Message *Sender::dequeMessage(){

        if (sendBuffer.empty()) {
                DEBUG(std::cout << "Sender Dequeueing NULL message" << std::endl;)
                return NULL;
        } else {
                DEBUG(std::cout << "Sender Dequeueing FULL message" << std::endl;)
                Message *m = sendBuffer.front();
                sendBuffer.pop_front();
                return m;
        }
}

void FastEcslent::Sender::runThread(){ // run the netManaager thread, private (each tick)

        try {
                socket = makeUDPBroadcastSocket();
                //socket->bind(udp::endpoint(boost::asio::ip::address_v4::broadcast( ), port));
                DEBUG(std::cout << "Sender Opened Broadcast Socket" << std::endl;)
                //udp::endpoint broadcastEndpoint(udp::v4(), port);
                udp::endpoint broadcastEndpoint(udp::endpoint(boost::asio::ip::address_v4::broadcast( ), port));
                //socket->bind(broadcastEndpoint);
                int size;
                while(!quit){

                        size_t len;
                        Message *m = dequeMessage();
                        DEBUG(std::cout << "Sender sending message at time: " << m->head.millisecondsFromStart << std::endl;)
                        while (m) {
                                size = (m->head.numberOfStructs * m->head.sizeOfStruct) + HeaderSize;
                                len = socket->send_to(boost::asio::buffer((void *) m, size), broadcastEndpoint);
                                DEBUG(std::cout << "Sender sent message with len: " << len << " and with size: " << size <<std::endl;)
                                delete m;
                                m = dequeMessage();
                        }
                        boost::this_thread::sleep(*sleepTime);
                }//end while
                } catch (std::exception& e) {
                        DEBUG(std::cerr << "Sender exception: " << e.what() << std::endl;)
                        quit = true;
                }
                DEBUG(std::cout << "..........................................Sender Dead......................................" << std::endl;)
}

void FastEcslent::Sender::stop(){ // end thread, public
        DEBUG(std::cout << "Sender: Stopping.....:" << senderThread.get_id() << std::endl;)
        boost::mutex::scoped_lock scoped_lock(quitLock);
        quit = true;
}

void FastEcslent::Sender::kill(){ // end thread, public
        DEBUG(std::cout << "Sender: Killed.....:" << std::endl;)
        quit = true;
        senderThread.interrupt();
}

void FastEcslent::Sender::join(){ //wait for thread to die, called by thread creator, public
        DEBUG(std::cout << "Sender: Shutting down socket" << std::endl;)
        boost::system::error_code ec;
        try {
                //socket->shutdown(udp::socket::shutdown_both);
                socket->close();
                //socket->close(ec);
                DEBUG(std::cout << "Sender STOPPING, Sender socket closing" << std::endl;)
                if (ec){
                        DEBUG(std::cerr << "Error in closing Sender socket:" << ec.message() << std::endl;)
                }
        } catch (std::exception& e) {
                DEBUG(std::cerr << "Exception: in closing Sender socket:" << e.what() << std::endl;)
        }
        DEBUG(std::cout << "Sender: Socket has been shutdown...waiting for join().....:" << std::endl;)
        std::flush(std::cout);
        senderThread.join();

}

---