Boost Asio 实现的UDP异步监听和TCP异步发送程序

 

#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/io_service.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/read_until.hpp>
#include <boost/asio/streambuf.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <iostream>
#include "Message.hpp"
#define MAXLEN 1024
#define SECONDS 5

using namespace std;
using namespace ****;
using namespace ****::Protocol;
using boost::asio::deadline_timer;
using boost::asio::ip::tcp;
using boost::asio::ip::udp;

class Session{
    public:
        Session(boost::asio::io_service& io_service, const char *path, int port)
            :stopped_(false),
            socket_(io_service),
            command_(io_service, udp::endpoint(udp::v4(), port)),
            deadline_(io_service),
            heartbeat_timer_(io_service)
    {
                callback_ = new Callback(path);
                input_stream_ = new char[MAXLEN];
            }
        ~Session(){
            delete callback_;
            delete input_stream_;
        }

        void start(tcp::resolver::iterator endpoint_iter){
            start_connect(endpoint_iter);
            deadline_.async_wait(boost::bind(&Session::check_deadline, this));
        }

        void stop(){
            stopped_ = true;
            socket_.close();
            deadline_.cancel();
            heartbeat_timer_.cancel();
        }


    private:
        void start_connect(tcp::resolver::iterator endpoint_iter){
            if(endpoint_iter != tcp::resolver::iterator()){
                logs.write_log(NORMAL, "inof:Connect--->");

                deadline_.expires_from_now(boost::posix_time::seconds(60));
                socket_.async_connect(endpoint_iter->endpoint(),
                        boost::bind(&Session::handle_connect,
                            this, _1, endpoint_iter));
            }
            else{
                stop();
            }
        }

        void handle_connect(const boost::system::error_code& ec,
                tcp::resolver::iterator endpoint_iter){

            if(stopped_)
                return;
            if(!socket_.is_open()){
                logs.write_log(NORMAL, "info:Connect time out");
                start_connect(++endpoint_iter);
            }
            else if(ec){
                logs.write_log(NORMAL, "info:Connect error");
                socket_.close();

                start_connect(++endpoint_iter);
            }
            else{
                //logs.write_log(NORMAL, "info:Conected to %s ", endpoint_iter->endpoint());

                logs.write_log(NORMAL, "info:Conected! ");
                start_read();
                start_write_heartbeat();
                start_delivery();
            }
        }

        void start_read(){
            deadline_.expires_from_now(boost::posix_time::seconds(SECONDS*6));
            socket_.async_receive(
                    boost::asio::buffer(&header_, HEADERLEN),
                    boost::bind(&Session::handle_read_header, this,
                        boost::asio::placeholders::error)
                    );
        }

        void handle_read_header(const boost::system::error_code& ec){
            if(stopped_)
                return;
            if(!ec){
                if(HEARTBEAT == header_.type){
                    logs.write_log(NORMAL, "info:Receive heartbeat message--->");
                    deadline_.expires_from_now(boost::posix_time::seconds(SECONDS*6));
                    //heartbeat_timer_.expires_from_now(boost::posix_time::seconds(SECONDS));

                    start_read();
                }
                else if(PACKET == header_.type){
                    logs.write_log(NORMAL, "info:Receive packet message--->");
                    socket_.async_receive(
                            boost::asio::buffer(input_stream_, header_.length),
                            boost::bind(&Session::handle_read_body, this,
                                boost::asio::placeholders::error)
                            );
                    logs.write_log(NORMAL, "info:Receive %d bytes", header_.length);
                }
            }
            else{
                logs.write_log(NORMAL, "Error on receive: %s", (ec.message()).c_str());
                stop();
            }
        }

        void handle_read_body(const boost::system::error_code& ec){
            if(stopped_)
                return;
            if(!ec){
                if(packet_message_.Parse(input_stream_, header_.length)){
                    Packet * packet = packet_message_.pack();
                    logs.write_log(NORMAL, "info:Receive data: %s", (packet->data()).c_str());
                    callback_->Write(packet);
                }

                start_read();
            }
            else{
                logs.write_log(NORMAL, "Error on receive: %s", (ec.message()).c_str());
                stop();
            }
        }

        void start_write_heartbeat(){
            if(stopped_)
                return ;
            Header header;
            header.length = 0;
            header.type = HEARTBEAT;
            boost::asio::async_write(socket_, boost::asio::buffer(&header, HEADERLEN),
                    boost::bind(&Session::handle_write_heartbeat, this,
                        boost::asio::placeholders::error    ));
            logs.write_log(NORMAL, "info:Send heartbeat message--->");
        }

        void handle_write_heartbeat(const boost::system::error_code& ec){
            if(stopped_)
                return ;

            if(!ec){
                heartbeat_timer_.expires_from_now(boost::posix_time::seconds(SECONDS));
                heartbeat_timer_.async_wait(boost::bind(&Session::start_write_heartbeat, this));

            }
            else{
                logs.write_log(NORMAL, "info:Error on heartbeat: %s ", (ec.message()).c_str());
                stop();

            }
        }

        void start_delivery(){
            if(stopped_)
                return ;
            Configure * conf = conf_message_.conf();
            Header header;
            conf->set_ip("10.14.2.40");
            conf->set_conf("\\d+\\w+*");
            conf->set_file("/tmp/aaa.txt");
            conf->set_chars("hello world");
            conf->set_offset(0);
            conf_message_.Serialize();
            header.length = conf_message_.length();
            header.type = CONFIG;
            socket_.async_send(
                    boost::asio::buffer(&header,    HEADERLEN),
                    boost::bind(&Session::handle_write_config, this,
                        boost::asio::placeholders::error)
                    );
        }

        void handle_write_config(const boost::system::error_code& ec){
            if(stopped_)
                return ;

            if(!ec){
                socket_.async_send(
                        boost::asio::buffer(conf_message_.data(),
                            conf_message_.length()),
                        boost::bind(&Session::handle_wait_commander, this,
                            boost::asio::placeholders::error)
                        );
            }
        }

        void handle_wait_commander(const boost::system::error_code& ec){
            logs.write_log(NORMAL, "info:Waiting for the commander --->..");
            /*
             *boost::bind(&Session::delivery--->.)
             * */
            command_.async_receive_from(
                    boost::asio::buffer(&commander_header_, HEADERLEN),
                    receive_entpoint_,
                    boost::bind(&Session::handle_check_commander, this,
                        boost::asio::placeholders::error)
                    );
        }

        void handle_check_commander(const boost::system::error_code& ec){
            if(DELIVERY == commander_header_.type || 8888 == commander_header_.length){
                logs.write_log(NORMAL, "info:Receive a Command, delivery a Configure ");
                /*
                boost::bind(&Session::start_delivery, this,
                        boost::asio::placeholders::error);
                        */
                start_delivery();
            }
            else{
                logs.write_log(NORMAL, "Error:Bad command format!");
            }
        }

        void check_deadline()
        {
            if (stopped_)
                return;

            // Check whether the deadline has passed. We compare the deadline against

            // the current time since a new asynchronous operation may have moved the

            // deadline before this actor had a chance to run.

            if (deadline_.expires_at() <= deadline_timer::traits_type::now())
            {
                // The deadline has passed. The socket is closed so that any outstanding

                // asynchronous operations are cancelled.

                socket_.close();

                // There is no longer an active deadline. The expiry is set to positive

                // infinity so that the actor takes no action until a new deadline is set.

                deadline_.expires_at(boost::posix_time::pos_infin);
            }

            // Put the actor back to sleep.

            deadline_.async_wait(boost::bind(&Session::check_deadline, this));
        }

    private:
        bool stopped_;
        tcp::socket socket_;
        udp::socket command_;
        udp::endpoint receive_entpoint_;
        Header header_;
        Header commander_header_;
        ConfigMessage conf_message_;
        DataMessage packet_message_;
        HeartbeatMessage heartbeat_message_;
        deadline_timer deadline_;
        deadline_timer heartbeat_timer_;
        Callback *callback_;
        char *input_stream_;

};

int main(int argc, char* argv[])
{
    try
    {
        if (argc != 4)
        {
            std::cerr << "Usage: Session   \n";
            return 1;
        }

        boost::asio::io_service io_service;
        tcp::resolver r(io_service);
        const char * path = "/home/lidawei/Doc/file";
        Session s(io_service, path, std::atoi(argv[3]));
        logs.write_log(NORMAL, "record file: %s", path);

        s.start(r.resolve(tcp::resolver::query(argv[1], argv[2])));
        logs.write_log(NORMAL, "info:Session start--->");

        io_service.run();
    }
    catch (std::exception& e)
    {
        std::cerr << "Exception: " << e.what() << "\n";
    }

    return 0;
}

g++ -Wall -g -I /usr/include/boost -L /usr/lib/boost easyzlib.o LogServer.pb.cc Message.cpp daemon.cpp Client.cpp -o Client -lboost_thread -lboost_system `pkg-config --cflags --libs protobuf`