使用 Boost asio 接收命令并执行它们

Question

我正在尝试制作一个增强服务器，它将接收命令并执行某些操作。现在我想创建一个函数，它将接收一个文件并将其保存到特定位置。问题在于序列化。我不知道如何以有效的方式识别流中的命令。我试过 boost::asio::read_until。实际上我的代码有效。第一个文件正在完美发送和接收。但是当客户端发送第二个文件时出现错误（提供的文件句柄无效）。我将非常感谢您的每一个建议。提前致谢！

    bool Sync::start_server() {



    boost::asio::streambuf request_buf; 
    std::istream request_stream(&request_buf);
    boost::system::error_code error; 


    try {

        tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), conf->def_port));
        acceptor.accept(socket); //socket is a member of class Sync
        while (true)
        {

            error.clear();


            size_t siz = boost::asio::read_until(socket, request_buf, "\n\n");

            std::cout << "request size:" << request_buf.size() << "\n";

            string command;
            string parameter;
            size_t data_size = 0;

            request_stream >> command;
            request_stream >> parameter;
            request_stream >> data_size;


            request_buf.consume(siz);//And also this

            //cut filename from path below
            size_t pos = parameter.find_last_of('\');
            if (pos != std::string::npos)
                parameter = parameter.substr(pos + 1);
            //cut filename from path above
            //command = "save";// constant until I make up other functions
             //execute(command, parameter, data_size);

            save(parameter,data_size);//parameter is filename

        }

    }
    catch (std::exception& e)
    {
        std::cerr << e.what() << std::endl;
    }
}

以及将文件保存到硬盘的功能：

bool Sync::save(string filename, size_t filesize) {
    boost::array<char, 1024> buf;
    cout << "filesize is" << filesize;
    size_t data_size = 0;
    boost::system::error_code error;
    std::ofstream output_file(filename.c_str(), std::ios_base::binary);
    if (!output_file)
    {
        std::cout << "failed to open " << filename << std::endl;
        return __LINE__;
    }
    while (true) {



        size_t len = socket.read_some(boost::asio::buffer(buf), error);

        if (len>0)
            output_file.write(buf.c_array(), (std::streamsize)len);
        if (output_file.tellp() == (std::fstream::pos_type)(std::streamsize)filesize)
        {

            output_file.close();
            buf.empty();
            break; // file was received
        }


        if (error)
        {
            socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error);
            socket.close(error);
            output_file.close();
            buf.empty();
            break;//an error occured
        }



    }


}

Answer 1

read_until 可能会超出定界符（因此 request_buf.size() 可以超过 siz）。当您实现 save 时，这是一个概念性问题，因为您从套接字读取 data_size 字节，这会忽略 request_buf

这些东西是代码味道：

if (output_file.tellp() == (std::fstream::pos_type)(std::streamsize)filesize) {

（从不使用 C-style 转换）。并且

return __LINE__; // huh? just `true` then

和

    buf.empty();

（这没有任何作用）。

我在这里展示三个版本：

第一次清理
简化（使用 tcp::iostream）
简化！（假设有关请求格式的更多信息）

第一次清理

这是一个合理的清理：

Live On Coliru

#include <boost/asio.hpp>
#include <boost/array.hpp>
#include <iostream>
#include <fstream>
namespace ba = boost::asio;
using ba::ip::tcp;

struct Conf {
    int def_port = 6767;
} s_config;

struct Request {
    std::string command;
    std::string parameter;
    std::size_t data_size = 0;

    std::string get_filename() const {
        // cut filename from path - TODO use boost::filesystem::path instead
        return parameter.substr(parameter.find_last_of('\') + 1);
    }

    friend std::istream& operator>>(std::istream& is, Request& req) {
        return is >> req.command >> req.parameter >> req.data_size;
    }
};

struct Sync {
    bool start_server();
    bool save(Request const& req, boost::asio::streambuf& request_buf);

    ba::io_service& io_service;
    tcp::socket socket{ io_service };
    Conf const *conf = &s_config;
};

bool Sync::start_server() {

    boost::asio::streambuf request_buf;
    boost::system::error_code error;

    try {

        tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), conf->def_port));
        acceptor.accept(socket); // socket is a member of class Sync
        while (true) {

            error.clear();

            std::string req_txt;

            {
                char const* delim = "\n\n";
                size_t siz = boost::asio::read_until(socket, request_buf, delim, error);

                // correct for actual request siz
                auto b = buffers_begin(request_buf.data()), 
                     e = buffers_end(request_buf.data());

                auto where = std::search(b, e, delim, delim+strlen(delim));
                siz = where==e
                    ? std::distance(b,e)
                    : std::distance(b,where)+strlen(delim);

                std::copy_n(b, siz, back_inserter(req_txt));
                request_buf.consume(siz); // consume only the request text bits from the buffer
            }

            std::cout << "request size:" << req_txt.size() << "\n";
            std::cout << "Request text: '" << req_txt << "'\n";
            Request req;

            {
                std::istringstream request_stream(req_txt);
                request_stream.exceptions(std::ios::failbit);
                request_stream >> req;
            }

            save(req, request_buf); // parameter is filename
        }

    } catch (std::exception &e) {
        std::cerr << "Error parsing request: " << e.what() << std::endl;
    }

    return false;
}

bool Sync::save(Request const& req, boost::asio::streambuf& request_buf) {
    auto filesize = req.data_size;
    std::cout << "filesize is: " << filesize << "\n";

    {
        std::ofstream output_file(req.get_filename(), std::ios::binary);
        if (!output_file) {
            std::cout << "failed to open " << req.get_filename() << std::endl;
            return true;
        }

        // deplete request_buf
        if (request_buf.size()) {
            if (request_buf.size() < filesize)
            {
                filesize -= request_buf.size();
                output_file << &request_buf;
            }
            else {
                // copy only filesize already available bytes
                std::copy_n(std::istreambuf_iterator<char>(&request_buf), filesize, 
                        std::ostreambuf_iterator<char>(output_file));
                filesize = 0;
            }
        }

        while (filesize) {
            boost::array<char, 1024> buf;
            boost::system::error_code error;

            std::streamsize len = socket.read_some(boost::asio::buffer(buf), error);

            if (len > 0)
            {
                output_file.write(buf.c_array(), len);
                filesize -= len;
            }

            if (error) {
                socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error); // ignore error
                socket.close(error);
                break; // an error occured
            }
        }
    } // closes output_file

    return false;
}

int main() {
    ba::io_service svc;

    Sync s{svc};
    s.start_server();

    svc.run();
}

与 echo -ne "save test.txt 12\n\nHello world\n" | netcat 127.0.0.1 6767:

这样的客户一起打印

request size:18
Request text: 'save test.txt 12

'
filesize is: 12
request size:1
Request text: '
'
Error parsing request: basic_ios::clear: iostream error

简化

然而，既然一切都是同步的，为什么不直接使用 tcp::iostream socket;。这将使 start_server 看起来像这样：

tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), conf->def_port));
acceptor.accept(*socket.rdbuf());

while (socket) {
    std::string req_txt, line;

    while (getline(socket, line) && !line.empty()) {
        req_txt += line + "\n";
    }

    std::cout << "request size:" << req_txt.size() << "\n";
    std::cout << "Request text: '" << req_txt << "'\n";

    Request req;
    if (std::istringstream(req_txt) >> req)
        save(req);
}

而save更简单：

void Sync::save(Request const& req) {
    char buf[1024];
    size_t remain = req.data_size, n = 0;

    for (std::ofstream of(req.get_filename(), std::ios::binary);
        socket.read(buf, std::min(sizeof(buf), remain)), (n = socket.gcount());
        remain -= n)
    {
        if (!of.write(buf, n))
            break;
    }
}

看到了Live On Coliru

测试时

for f in test{a..z}.txt; do (echo -ne "save $f 12\n\nHello world\n"); done | netcat 127.0.0.1 6767

打印：

request size:18
Request text: 'save testa.txt 12
'
request size:18
Request text: 'save testb.txt 12
'
[... snip ...]
request size:18
Request text: 'save testz.txt 12
'
request size:0
Request text: ''

更简单

如果您知道请求是单行的，或者空格不重要：

struct Sync {
    void run_server();
    void save(Request const& req);

  private:
    Conf const *conf = &s_config;
    tcp::iostream socket;
};

void Sync::run_server() {
    ba::io_service io_service;
    tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), conf->def_port));
    acceptor.accept(*socket.rdbuf());

    for (Request req; socket >> std::noskipws >> req; std::cout << req << " handled\n")
        save(req);
}

void Sync::save(Request const& req) {
    char buf[1024];
    size_t remain = req.data_size, n = 0;

    for (std::ofstream of(req.get_filename(), std::ios::binary);
        socket.read(buf, std::min(sizeof(buf), remain)), (n = socket.gcount());
        remain -= n) 
    {
        if (!of.write(buf, n)) break;
    }
}

int main() {
    Sync().run_server();
}

这就是大约 33 行代码的整个程序。看到了Live On Coliru,打印:

Request {"save" "testa.txt"} handled
Request {"save" "testb.txt"} handled
Request {"save" "testc.txt"} handled
[... snip ...]
Request {"save" "testy.txt"} handled
Request {"save" "testz.txt"} handled

使用 Boost asio 接收命令并执行它们

Using Boost asio to receive commands and execute them

c++

boost

istream

streambuf

boost-asio

第一次清理

简化

更简单