断言失败：(id_!= T::id)，函数 try_lock，文件 /usr/local/include/boost/beast/websocket/detail/stream_base.hpp，第 91 行。
// Echoes back all received WebSocket messages
class session : public std::enable_shared_from_this<session>
{
websocket::stream<tcp::socket> ws_;
boost::asio::strand<
boost::asio::io_context::executor_type> strand_;
boost::beast::multi_buffer buffer_;
public:
// Take ownership of the socket
explicit
session(tcp::socket socket)
: ws_(std::move(socket))
, strand_(ws_.get_executor())
{
}
// Start the asynchronous operation
void
run()
{
// Accept the websocket handshake
ws_.async_accept(
boost::asio::bind_executor(
strand_,
std::bind(
&session::on_accept,
shared_from_this(),
std::placeholders::_1)));
}
void
on_accept(boost::system::error_code ec)
{
std::cout << std::this_thread::get_id() <<" : DO ACCEPT" << std::endl;
if(ec)
return fail(ec, "accept");
std::ifstream ifs("tweetsample.txt");
if(ifs.good())
{
auto tweet = std::string{};
while(std::getline(ifs,tweet))
{
try
{
std::this_thread::sleep_for(std::chrono::seconds(1));
auto t = json::parse(tweet);
std::string tweet_text = t["text"];
auto n = boost::asio::buffer_copy(buffer_.prepare(tweet_text.size()),
boost::asio::buffer(tweet_text));
buffer_.commit(n);
do_write();
}
catch(nlohmann::detail::type_error& ex)
{
std::cout << ex.what() << std::endl;
}
}
}
}
void
do_read()
{
// Read a message into our buffer
ws_.async_read(
buffer_,
boost::asio::bind_executor(
strand_,
std::bind(
&session::on_read,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
void do_write()
{
std::cout << std::this_thread::get_id() << "do_write" << std::endl;
ws_.async_write(
buffer_.data(),
boost::asio::bind_executor(
strand_,
std::bind(
&session::on_write,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
void
on_read(
boost::system::error_code ec,
std::size_t bytes_transferred)
{
boost::ignore_unused(bytes_transferred);
// This indicates that the session was closed
if(ec == websocket::error::closed)
return;
if(ec)
fail(ec, "read");
std::cout << boost::beast::buffers(buffer_.data()) << std::endl;
// Echo the message
ws_.text(ws_.got_text());
ws_.async_write(
buffer_.data(),
boost::asio::bind_executor(
strand_,
std::bind(
&session::on_write,
shared_from_this(),
std::placeholders::_1,
std::placeholders::_2)));
}
void
on_write(
boost::system::error_code ec,
std::size_t bytes_transferred)
{
std::cout << "on_write" << std::endl;
boost::ignore_unused(bytes_transferred);
if(ec)
return fail(ec, "write");
// Clear the buffer
buffer_.consume(buffer_.size());
// Do another read
do_read();
}
};
//------------------------------------------------------------------------------
// Accepts incoming connections and launches the sessions
class listener : public std::enable_shared_from_this<listener>
{
tcp::acceptor acceptor_;
tcp::socket socket_;
public:
listener(
boost::asio::io_context& ioc,
tcp::endpoint endpoint)
: acceptor_(ioc)
, socket_(ioc)
{
boost::system::error_code ec;
// Open the acceptor
acceptor_.open(endpoint.protocol(), ec);
if(ec)
{
fail(ec, "open");
return;
}
// Allow address reuse
acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
if(ec)
{
fail(ec, "set_option");
return;
}
// Bind to the server address
acceptor_.bind(endpoint, ec);
if(ec)
{
fail(ec, "bind");
return;
}
// Start listening for connections
acceptor_.listen(
boost::asio::socket_base::max_listen_connections, ec);
if(ec)
{
fail(ec, "listen");
return;
}
}
// Start accepting incoming connections
void
run()
{
if(! acceptor_.is_open())
return;
do_accept();
}
void
do_accept()
{
acceptor_.async_accept(
socket_,
std::bind(
&listener::on_accept,
shared_from_this(),
std::placeholders::_1));
}
void
on_accept(boost::system::error_code ec)
{
if(ec)
{
fail(ec, "accept");
}
else
{
// Create the session and run it
std::make_shared<session>(std::move(socket_))->run();
}
// Accept another connection
do_accept();
}
};
int main(int argc, char* argv[])
{
auto const address = boost::asio::ip::make_address("XXX.XX.XX.X");
auto const port = static_cast<unsigned short>(std::atoi("XXXX"));
auto const threads = 1;
// The io_context is required for all I/O
boost::asio::io_context ioc{threads};
// Create and launch a listening port
std::make_shared<listener>(ioc, tcp::endpoint{address, port})->run();
// Run the I/O service on the requested number of threads
std::vector<std::thread> v;
v.reserve(threads - 1);
for(auto i = threads - 1; i > 0; --i)
v.emplace_back(
[&ioc]
{
ioc.run();
});
ioc.run();
return EXIT_SUCCESS;
}
我正在尝试使用 websocket 进行试验，目前当我接受接受时，我想逐行发送我从文件中读取的所有数据。
发送第一行后程序崩溃。请告知我做错了什么。