Ticket #13477: socket.cpp

#define _CRT_SECURE_NO_WARNINGS
#include <ctime>
#include <iostream>
#include <string>
#include <queue>
#include <boost/array.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/thread.hpp>
#include <boost/thread/thread.hpp> 
#include <boost/chrono.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

using boost::asio::ip::udp;
using std::cout;
using std::cin;
using std::endl;
using std::string;
using namespace std;

template<typename T>
std::string toString(const T& value);
std::string IntToString(const int& i);

class UdpCore
{
private:
        boost::asio::ip::udp::endpoint endpoint;
        boost::asio::ip::udp::socket socket;

        string multicast_address;
        unsigned short multicast_port;
        boost::thread_group threads;    // thread group
        boost::thread* thread_main;             // main thread
        boost::thread* thread_listen;   // listen thread
        boost::thread* thread_getsend;  // get/send thread
        boost::mutex stopMutex;
        bool initialize = false;
        bool stop, showBroadcast;
        int i_getsend, i_listen, i_main, i_message, interval;
        string message;
public:
        // constructor
        UdpCore(boost::asio::io_service& io_service, std::string multicast_address, unsigned short multicast_port, int interval, bool show = false)
                : endpoint(boost::asio::ip::address::from_string(multicast_address), multicast_port),
                socket(io_service, endpoint.protocol()),
                multicast_address(multicast_address),
                multicast_port(multicast_port),
                interval(interval),
                showBroadcast(show)
        {
                Initialize(io_service, show);
        }

        ~UdpCore()
        {
                // show exit message
                cout << "Exiting UDP Core." << endl;
        }

        // initialize
        void Initialize(boost::asio::io_service& io_service, bool show = false)
        {
                if (initialize == false)
                {
                        GetInfo();
                }

                boost::asio::ip::udp::endpoint endpoint(boost::asio::ip::make_address(multicast_address), multicast_port);
                boost::asio::ip::udp::socket socket(io_service, endpoint.protocol());
                socket.set_option(boost::asio::ip::udp::socket::reuse_address(true));   // no need

                thread_main = new boost::thread(boost::ref(*this));
                thread_getsend = new boost::thread(&UdpCore::Callable_GetSend, this, interval, boost::ref(i_listen), boost::ref(message));
                threads.add_thread(thread_getsend);     // get/send thread
                stop = false;
                showBroadcast = show;
                i_getsend = 0;
                i_listen = 0;
                i_main = 0;
                i_message = 0;
                message.clear();

                initialize = true;
        }

        void GetInfo()
        {
                multicast_address = "192.168.0.255";
                multicast_port = 13000;
                interval = 500;
        }

        // start the threads
        void Start()
        {
                // Wait till they are finished
                threads.join_all();
        }

        // stop the threads
        void Stop()
        {
                // warning message
                cout << "Stopping all threads." << endl;

                // signal the threads to stop (thread-safe)
                stopMutex.lock();
                stop = true;
                stopMutex.unlock();

                // wait for the threads to finish
                thread_main->interrupt();       // in case not interrupted by operator()
                threads.interrupt_all();
                threads.join_all();

                // close socket after everything closes
                socket.close();
        }

        void Callable_Listen(int interval, int& count)
        {
                while (!stop)
                {
                        if (message != "")
                                socket.async_send_to(boost::asio::buffer(message), endpoint, [this](boost::system::error_code ec, std::size_t /*length*/)
                        {
                                stopMutex.lock();
                                if (showBroadcast)
                                {
                                        cout << i_message << " - " << message << endl;  // show  count
                                }
                                message.clear();        //clear after sending
                                stopMutex.unlock();
                        });
                        ++i_message;

                        // wait routine
                        boost::this_thread::sleep(boost::posix_time::millisec(interval));
                        boost::this_thread::interruption_point();
                        ++i_listen;
                }
        }

        void Callable_GetSend(int interval, int& count, string& userInput)
        {
                while (!stop)
                {
                        stopMutex.lock();
                        cout << "Callable_GetSend [" << count++ << "]. Enter message: ";
                        getline(cin, userInput);
                        if (message != "")
                                socket.async_send_to(boost::asio::buffer(message), endpoint, [this](boost::system::error_code ec, std::size_t /*length*/)
                        {
                                if (showBroadcast)
                                {
                                        cout << i_message << " - " << message << endl;  // show  count
                                }
                                message.clear();        //clear after sending
                        });
                        stopMutex.unlock();

                        // wait routine
                        boost::this_thread::sleep(boost::posix_time::millisec(interval));
                        boost::this_thread::interruption_point();
                        ++i_getsend;
                        ++i_message;
                }
        }

        // Thread function
        void operator () ()
        {
                while (!stop)
                {
                        if (message == "STOP")
                        {
                                try
                                {
                                        this->Stop();
                                }
                                catch (exception e)
                                {
                                        cout << e.what() << endl;
                                }
                        }

                        boost::this_thread::sleep(boost::posix_time::millisec(interval));
                        boost::this_thread::interruption_point();
                }
        }

        std::string make_daytime_string()
        {
                using namespace std; // For time_t, time and ctime;
                time_t now = time(0);
                std::string result = ctime(&now);
                return result.erase(result.length() - 1, 1);
        }

        std::string some_string()
        {
                std::string result;
                result = make_daytime_string();
                return result;
        }
};

int main()
{
        try
        {
                boost::asio::io_service io_service;
                UdpCore mt(io_service, "192.168.0.255", 13000, 5000, false);
                mt.Start();
        }
        catch (std::exception& e)
        {
                std::cerr << "Exception: " << e.what() << "\n";
        }
}