Communication.cpp

Go to the documentation of this file.

#include "Communication.h"

namespace castor { namespace net {

        boost::mutex Communication::mutex;
        CommunicationPtr Communication::instance;

        Communication::Communication() :
                threadPool(), service(), initialised(false)
        {
        }

        Communication::~Communication() {
        }

        CommunicationPtr Communication::getInstance() {

                boost::mutex::scoped_lock lock(mutex);

                if (!instance) {
                        instance = CommunicationPtr(new Communication());
                }

                return instance;
        }

        CastorChannelPtr Communication::create(const std::string spec) throw(NetException) {
                return CastorChannel::create(this->service, spec);
        }

        CastorChannelPtr Communication::create(const NetAddress &address) throw(NetException) {
                return CastorChannel::create(this->service, address);
        }

        void Communication::run() {
                std::cout << "geminga Communication run " << std::endl;
                this->service.run();
                std::cout << "geminga Communication stopped " << std::endl;
        }

        void Communication::start(size_t threads) {

                boost::mutex::scoped_lock lock(mutex);

                if (this->initialised) return;

                this->initialised = true;

                for (size_t i = 0; i < threads; i++) {
                        this->threadPool.push_back(ThreadPtr(new asio::thread(
                                        boost::bind(&Communication::run, this))));
                }

                initialised = true;
                std::cout << "Castor communication subsystem started (" << threads << " threads)" << std::endl;
        }

        void Communication::stop() {

                boost::mutex::scoped_lock lock(mutex);

                this->service.stop();

                for (ThreadPool::iterator itr = this->threadPool.begin();
                         itr != this->threadPool.end(); itr++)
                {
                        (*itr)->join();
                }

                this->threadPool.clear();

//              std::cout << "Castor communication subsystem stopped" << std::endl;
        }

        void Communication::wait() {
                std::cout << "Start join" << std::endl;
                this->threadPool[0]->join();
                std::cout << "Stop join" << std::endl;
        }

        asio::io_service &Communication::asioService() {
                return this->service;
        }

} }