TCPServer.cpp

Go to the documentation of this file.

#include "TCPServer.h"
#include "StockManager.h"

#include <iostream>
#include <istream>

#include <boost/bind.hpp>
#include <boost/asio/streambuf.hpp>


inline
tcp_connection::pointer tcp_connection::create(boost::asio::io_service& io_service) {
#ifdef DEBUG_CPP
        std::cout << "tcp_connection::create return a new tcp_connection" << std::endl;
#endif

    return pointer(new tcp_connection(io_service));
}

// to modify for my purpose, in RealTime case, nothing to send back ! or just ok no error !
void tcp_connection::start() {

#ifdef DEBUG_CPP
    std::cout << " tcp_connection::start() will associate handle_write" << std::endl;
#endif

    // make async. call and send callback to handle_write
    // not sure how working but response not used at this point
    boost::asio::async_write(socket_, boost::asio::buffer(message_),
        boost::bind(&tcp_connection::handle_write, shared_from_this(),
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));
}

// to send an answer
void tcp_connection::handle_write(const boost::system::error_code& error,
      size_t /*bytes_transferred*/)
{
#ifdef DEBUG_CPP
    std::cout << "Entry connection::handle_write, do nothing " << std::endl;
    //std::cout << "check error " << error << " size message back " << bytes_transferred << std::endl;
    // not working, message already sent... what the point, receive the answer ??
    //message_= "modified toto";
#endif

    if ( error ) {
        std::cout << "tcp_connection::handle_write get error " << error << std::endl;
    }
}



// Constructor with StockManager and call with io_service, deprecated
TCPServer::TCPServer ( StockManager *sm, unsigned int iport, boost::asio::io_service& io_service ) :
           f_sm( sm ), port( iport ), acceptor_( io_service, tcp::endpoint(tcp::v4(), iport)) {
#ifdef DEBUG_CPP
    std::cout << "Constructor TCPServer port with stockmanager " << port << " " << f_sm << std::endl;
#endif
    // default at creation
    is_running = false;
    want_exit = false;
    //t_ = std::thread{ &TCPServer::LoopThread, this };
}

// io_service created inside the function
TCPServer::TCPServer ( StockManager *sm, unsigned int iport ) :
           f_sm( sm ), port( iport ), my_io_service(), acceptor_( my_io_service, tcp::endpoint(tcp::v4(), iport)) {
#ifdef DEBUG_CPP
    std::cout << "Constructor TCPServer port with stockmanager " << port << " " << f_sm << std::endl;
#endif
    // do not compile
    //acceptor_ = tcp::acceptor(my_io_service, tcp::endpoint(tcp::v4(), port));

    // default at creation
    is_running = false;
    want_exit = false;
    //t_ = std::thread{ &TCPServer::LoopThread, this };
}

TCPServer::~TCPServer()
{
#ifdef DEBUG_CPP
    std::cout << "Destructor TCPServer " << std::endl;
#endif
    // not necessary in unit_test run_stop_miss_stop, but if active ??
    if ( is_running )
        StopServer();

    // important bracket, lock is deleted this way
    /*
    {
        std::unique_lock<std::mutex> lock( loop_mutex );
        want_exit = true;
        loop_cv.notify_one();
    }
    // wait termination of the thread
    t_.join();
    */
}


void TCPServer::StartServer()
{
#ifdef DEBUG_CPP
    std::cout << "Entry TCPServer::StartServer() TCPServer port " << port << std::endl;
#endif

    //is_running = true;
    t_ = std::thread{ &TCPServer::LoopThread, this };


    //acceptor_.open( my_io_service, tcp::endpoint(tcp::v4(), port));
    //acceptor_.open(); //tcp::endpoint(tcp::v4(), port));
    /* http://stackoverflow.com/questions/15829165/how-to-launch-an-event-when-my-boostasio-tcp-server-just-start-running-aka/15883510#15883510
     * http://stackoverflow.com/questions/15926077/boostasio-acceptor-reopen-and-async-read-after-eof
    using boost::asio::ip::tcp;
    tcp::endpoint endpoint_(tcp::v4(), 123);
    tcp::acceptor acceptor_(io_service); // closed state
    acceptor.open(endpoint_.protocol()); // opened state
    acceptor.bind(endpoint);
    acceptor.listen();
    */
    /*
    std::cout << "test is open " << acceptor_.is_open() << std::endl;
    if ( acceptor_.is_open() == false  ) {
        tcp::endpoint endpoint_(tcp::v4(), port);
        acceptor_.open( endpoint_.protocol() );
        //acceptor_.bind(endpoint_);
        //acceptor_.listen();
    }
    // recreate connection
    start_accept();
        */

    // here important mutex ?? bracket ??
    {
        std::unique_lock<std::mutex> lock( loop_mutex );
        is_running = true;

        // new added delete _connection, here need to recreate
        // signal condition variable has changed. equivalent notify_all() here
        loop_cv.notify_one();
    }

}

// allow to not have the TCP recreating when deleting
void TCPServer::StopServer() {
#ifdef DEBUG_CPP
        std::cout << "TCPServer::StopServer is_running " << is_running << std::endl;
#endif

        acceptor_.get_io_service().stop();
        // important bracket, lock is deleted this way
        {
                std::unique_lock<std::mutex> lock( loop_mutex );
                want_exit = true;
                loop_cv.notify_one();
        }
        // wait termination of the thread
        //t_.join();
        is_running = false;
        t_.join();

        // not sure if running or not
        /*
        if ( is_running ) {

                //{
                        std::unique_lock<std::mutex> lock( loop_mutex );
                        // update status
                        is_running = false;
                        loop_cv.notify_one();
                //}
        */
                // not promising...
                // not enought, tcp_connection still alive and recreated..
                // stopping io_service is not enough, maybe acceptor.listen / stop_listen
                // try to deal with connection
                //delete _connection.get();
                /*
                std::cout << "before reset() count() " << _connection.use_count() << std::endl;
                _connection.reset();
                std::cout << "after reset() count() " << _connection.use_count() << std::endl;
                */

                // necessary ?? yes otherwise service always running, cannot exit

                // thread will make an other loop
                // reset() must be called before run()
                // in fact it is received, but executed after the restart !!
                // need stop to quit from run()


                // This function is used to close the acceptor. Any asynchronous accept operations will be cancelled immediately.
                // A subsequent call to open() is required before the acceptor can again be used to again perform socket accept operations.
                //acceptor_.cancel();
                //acceptor_.close();
        //}

    // reset() must be called before run()
        //acceptor_.get_io_service().stop();
}

// thread created in constructor, enter infinite loop to keep the thread alive until deletion of the object
void TCPServer::LoopThread() {
#ifdef DEBUG_CPP
        std::cout << "TCPServer::LoopThread " << std::endl;
#endif

        // here create one initial tcp_connection, bound to handle_accept
        // commented if connection deal in Start/Stop server
        start_accept();

        while ( true ) {

                { //not sure important here, important to release the lock, stopserver blocked otherwise
                        // must aquire the lock first
                        std::unique_lock<std::mutex> lock( loop_mutex );
                        // usual call, no predicate
                        // wait(unique_lock<mutex>& lk);

                        std::cout << "TCPServer::LoopThread will wait a notify " << std::endl;
                        // while predicate is false, wait
                        // wait relax the lock, wait for a signal, relock and executes
                        loop_cv.wait( lock, [&](){ return is_running || want_exit ; } );
                        //loop_cv.wait( lock );

                        if ( want_exit ) {
                                std::cout << "LoopThread will exit and terminate" << std::endl;
                                return;
                        }

                }

                std::cout << "TCPServer::LoopThread() before acceptor_.get_io_service().run() " << std::endl;

                // blocking function, reset() solve the problem, must be called before a new run
                //acceptor_.get_io_service().reset();
            acceptor_.get_io_service().run();

            // will pass when get_io_service.stop() will be called
            //std::cout << "TCPServer::LoopThread() after acceptor_.get_io_service().run() " << std::endl;
        }


}

// create the connection, wait to receive a message async.
void TCPServer::start_accept() {

#ifdef DEBUG_CPP
    std::cout << "TCPServer::start_accept create connection & async_accept" << std::endl;
#endif

    // create a new connection
    tcp_connection::pointer new_connection = tcp_connection::create(acceptor_.get_io_service());
    //_connection =
    //    tcp_connection::create(acceptor_.get_io_service());


    // Verify socket is in a closed state. Private
    //new_connection->socket_.close();
    // no change, next execution call new_connection is close
    //new_connection->GetSocket().close();

    // use a non-blocking call and bind the received message to handle_accept
    acceptor_.async_accept( new_connection->socket(),
                boost::bind(&TCPServer::handle_accept, this, new_connection,
                        boost::asio::placeholders::error));
}

// new_connection is used to read the message (and answer)
// but it is automatically deleted at the end of the function. An other connection is created at the very end.
// f_sm->SendDataCSV() is run synchro. in this function (but this function is run async.)
void TCPServer::handle_accept(tcp_connection::pointer new_connection,
                   const boost::system::error_code& error)
{
#ifdef DEBUG_CPP
    std::cout << "TCPServer::handle_accept, new_connection->socket().is_open() " << new_connection->socket().is_open() << std::endl;
    std::cout << "is_running " << is_running << std::endl;
#endif

    boost::asio::streambuf message;
    std::string tmp_string;

    // what kind of error ??
    if ( error )
        std::cout << " TCPServer::handle_accept() error: " << error << std::endl;

    // fix delimiter " END" for the end of the message
    try {
        boost::asio::read_until( new_connection->socket(), message, " END");

//#ifdef DEBUG_CPP
      // do not compile
//    std::cout << "after read message: " << message.data() << std::endl;
//#endif

            // transform message to string, tricky, but from website
            std::istream is(&message);
            std::getline(is, tmp_string);
            // delete the end delimiter: " END"
            tmp_string.erase(tmp_string.size()-4);
//#ifdef DEBUG_CPP
//      std::cout << "TCPServer::handle_accept message_string!" << tmp_string << "!" << std::endl;
//#endif

            // call to StockManager function to extract the data and update the Stock
        // throw StockException if bad format
            std::cout << "before SendDataCSV" << std::endl;
            f_sm->SendDatadCSV( tmp_string );

    } catch (boost::system::system_error &e) {
        std::cout << "Got exception boost::system::system_error from boost:asio::read_until() " << e.what() << std::endl;

    // StockExcpetion if bad format, avoids crash, and start_accept is called again
    } catch (StockException &e ) {
        std::cout << "\nTCPServer, got a StockException " << e.what() << std::endl;
        // set error_code in the boost way
        // or throw C++11 exception:
        //  Would need to copy into the main thread. From here:
        // catch(..)
        //  err = current_exception()
        //  a SM::SetThreadError(exception_ptr &err) (in a list or queue), why not in RealTime ? use observer pattern 2 types of messages ?
        //     then main thread checks (Timer again) if some errors occurred
        // more general :

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

    // assign a non-blocking function to answer to the message, not used now
/*
    if (!error) {
#ifdef DEBUG_CPP
        std::cout << "TCPServer::handle_accept() run new_connection->start" << std::endl;
#endif
        // seems only to associate handle_write, which is not used.
        // if commented, working as well, certainly cannot send back data
        //new_connection->start();
    } else {
        std::cout << "Error in TCPServer::handle_accept " << error << std::endl;
    }
*/

    // restart a new connection for receiving data
//#ifdef DEBUG_CPP
//        std::cout << "TCPServer::handle_accept() run start_accept()" << std::endl;
//#endif
    start_accept();

#ifdef DEBUG_CPP
        std::cout << "End TCPServer::handle_accept()" << std::endl;
#endif
}