//跨平台异步http server
#define _WIN32_WINNT 0x0A00
#include <iostream>
#include <vector>
#include <string>
#include <ctime> //std::tm,std::strftime
#include <chrono>
#include <thread>
#include <mutex>
#include <functional>
#include <algorithm>#include "boost/asio/connect.hpp"
#include "boost/asio/io_context.hpp"
#include "boost/asio/ip/tcp.hpp"
#include "boost/asio/read.hpp"
#include "boost/asio/write.hpp"
#include "boost/asio/streambuf.hpp"
#include "boost/asio/read_until.hpp"
#include "boost/core/ignore_unused.hpp"
#include "boost/asio/strand.hpp"
#include "boost/asio/bind_executor.hpp"
#include "llhttp.h"
#include "boost/asio/ssl.hpp"using namespace boost;
namespace ssl = boost::asio::ssl;
namespace ip = boost::asio::ip;enum { BUF_SIZE = 1024 };
const int timeoutSeconds = 300;class HttpServer:public std::enable_shared_from_this<HttpServer>
{class Session;
public:HttpServer(boost::asio::io_context& ioc, uint16_t port) :ioc_(ioc), acceptor_(ioc, ip::tcp::endpoint(ip::tcp::v4(),port)),sslContext_(ssl::context::tlsv12_server){//httpssslContext_.use_certificate_chain_file(R"(C:\Users\Administrator\Desktop\test\cert\fullchain.pem)");sslContext_.use_private_key_file(R"(C:\Users\Administrator\Desktop\test\cert\privkey.pem)", ssl::context::pem);sslContext_.set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::single_dh_use);acceptor_.set_option(asio::ip::tcp::acceptor::reuse_address(true));//地址重用auto endpoint = acceptor_.local_endpoint();LogInfo() << "[acceptor] fd=" << acceptor_.native_handle() << ",ip=" << endpoint.address().to_string() << ",port=" << endpoint.port();acceptor_.listen();//启动监听}void Start(){DoAccept();}void RegisterHandler(std::string url, std::function<void(std::string&,bool&)> handler){handlers_[url] = handler;}// 时间串转换函数static std::string getCurrentTimeFormatted(){auto now = std::chrono::system_clock::now();std::time_t now_time = std::chrono::system_clock::to_time_t(now);std::tm tm_time;#if defined(_WIN32)localtime_s(&tm_time, &now_time);#elselocaltime_r(&now_time, &tm_time);#endifstd::ostringstream oss;oss << std::put_time(&tm_time, "%Y-%m-%d %H:%M:%S");return oss.str();}class LogInfo{public:LogInfo() { info_.reserve(256); }template<typename T>LogInfo& operator<<(T&& info){oss_ << std::forward<T>(info);return *this;}~LogInfo() { std::lock_guard<std::mutex> lg(mtx_);std::cout << "[" << HttpServer::getCurrentTimeFormatted() << "]: " << oss_.str() << "\n"; }private:std::string info_;std::ostringstream oss_;static std::mutex mtx_;};class LogErr{public:LogErr() { err_.reserve(256); }template <typename T>LogErr& operator<<(T&& err){oss_ << std::forward<T>(err);return *this;}~LogErr() { std::lock_guard<std::mutex> lg(mtx_);std::cerr << "[" << HttpServer::getCurrentTimeFormatted() << "]: " << oss_.str() << "\n"; }private:std::string err_;std::ostringstream oss_;static std::mutex mtx_;};private:void DoAccept(){auto self = shared_from_this();//acceptor_.async_accept([this, self](boost::system::error_code errCode, ip::tcp::socket socket) {auto sptrNewSocket = std::make_shared<ssl::stream<ip::tcp::socket>>(ioc_, sslContext_);acceptor_.async_accept(sptrNewSocket->next_layer(), [this, self, sptrNewSocket](boost::system::error_code errCode) {if (!errCode){sptrNewSocket->async_handshake(ssl::stream_base::server, [this, self, sptrNewSocket](boost::system::error_code errCode) {if (!errCode){LogInfo() << "[DoAccept] handshake ok, start read..";//新建会话auto sptrStrand = std::make_shared<boost::asio::strand<boost::asio::io_context::executor_type>>(boost::asio::make_strand(ioc_));auto sptrSession = std::make_shared<Session>(self->ioc_, sptrNewSocket, sptrStrand);//读取到连续2个换行回车后，判定请求行+请求头读取完sptrSession->StartTimer(timeoutSeconds);asio::async_read_until(*(sptrSession->sptrSocket_), *(sptrSession->sptrBuffer_), "\r\n\r\n", boost::asio::bind_executor(*sptrStrand, [this, sptrSession](boost::system::error_code errCode, std::size_t headerLength) {sptrSession->CancelTimer();LogInfo() << "[accept] " << (errCode ? errCode.message() : "new client..");LogInfo() << "[DoAccept] async_read_until: " << headerLength << " bytes..";//若连接被对端client关闭，直接调用关闭if (errCode == asio::error::eof || errCode == asio::error::connection_reset){LogInfo() << "[DoAccept] client closed connection during read..";sptrSession->CloseSession();return;}if (!errCode){//使用llhttp解析http报文sptrSession->ParseRequest(handlers_);}else{sptrSession->CloseSession();}}));}});}DoAccept();});}private:boost::asio::io_context ioc_;ssl::context sslContext_;ip::tcp::acceptor acceptor_;std::unordered_map<std::string, std::function<void(std::string&, bool&)>> handlers_;class Session : public std::enable_shared_from_this<Session>{public://Session(boost::asio::io_context& ioc, ip::tcp::socket socket, std::shared_ptr<boost::asio::strand<boost::asio::io_context::executor_type>> sptrStrand, bool keep_alive = true)Session(boost::asio::io_context& ioc, std::shared_ptr<ssl::stream<ip::tcp::socket>> socket, std::shared_ptr<boost::asio::strand<boost::asio::io_context::executor_type>> sptrStrand, bool keep_alive=true):sptrSocket_(socket),keepAlive_(keep_alive),timer_(ioc){ sptrBuffer_ = std::make_shared<asio::streambuf>(); sptrStrand_ = sptrStrand;InitLLHttp();}void InitLLHttp(){llhttp_settings_init(&Session::settings_);llhttp_init(&parser_, HTTP_BOTH, &Session::settings_);parser_.data = this; //存一下this，在回调里用Session::settings_.on_url = [](llhttp_t* ptrParser, const char* at, size_t length) {auto session = static_cast<Session*>(ptrParser->data);session->url_.assign(at, length);return 0;};Session::settings_.on_header_field = [](llhttp_t* ptrParser, const char* at, size_t length) {auto session = static_cast<Session*>(ptrParser->data);session->currentHeaderField_.assign(at, length);return 0;};Session::settings_.on_header_value = [](llhttp_t* ptrParser, const char* at, size_t length) {auto session = static_cast<Session*>(ptrParser->data);session->headers_[session->currentHeaderField_] = std::string(at, length);return 0;};Session::settings_.on_body = [](llhttp_t* ptrParser, const char* at, size_t length) {auto session = static_cast<Session*>(ptrParser->data);session->body_.assign(at, length);return 0;};Session::settings_.on_message_complete = [](llhttp_t* ptrParser) {auto session = static_cast<Session*>(ptrParser->data);session->isParseComplete_ = true;return 0;};}//读取http体，并通过llhttp解析void ParseRequest(std::unordered_map<std::string,std::function<void(std::string&, bool&)>>& handlers){if (isParseComplete_){Reset();}std::string httpFrame(boost::asio::buffers_begin(sptrBuffer_->data()), boost::asio::buffers_end(sptrBuffer_->data()));LogInfo() << "[ParseRequest] got " << httpFrame.size() << " bytes\n" << httpFrame;llhttp_errno_t err = llhttp_execute(&parser_, httpFrame.data(), httpFrame.size());if (isParseComplete_){LogInfo() << "[ParseRequest] HttpFrame complete,url: " << url_;ProcessRequestBody(handlers);//缓冲区中已被解析的数据需清除sptrBuffer_->consume(httpFrame.size());}else{LogInfo() << "[ParseRequest] HttpFrame not complete, need read again..";ReadRequestBody(handlers);}}//读取body void ReadRequestBody(std::unordered_map<std::string,std::function<void(std::string&, bool&)>>& handlers){if (headers_.count("Content-Length") == 0){SendErrorResponse(404, "Not Found", handlers);return;}std::size_t content_length = std::stoul(headers_.at("Content-Length"));//改http请求的body长度std::size_t receivedLength = sptrBuffer_->size(); //当前缓冲区里有的body长度if (receivedLength >= content_length){//body数据足够，处理ProcessRequestBody(handlers);}else{//读取缺少的bodyStartTimer(timeoutSeconds);boost::asio::async_read(*sptrSocket_, *sptrBuffer_, asio::transfer_exactly(content_length-receivedLength), [self = shared_from_this(), &handlers](const boost::system::error_code errCode, std::size_t) {self->CancelTimer();self->ParseRequest(handlers);});}}//处理bodyvoid ProcessRequestBody(std::unordered_map<std::string,std::function<void(std::string&, bool&)>>& handlers){auto handler = handlers.find(url_);//iteratorif (handler == handlers.end()){SendErrorResponse(404, "Not Found", handlers);return;}this->keepAlive_ = true;if (headers_.count("Connection") != 0 && headers_.at("Connection") == "close")this->keepAlive_ = false;auto response = std::make_shared<std::string>();response->reserve(1024 * 5);handler->second(*response, keepAlive_);SendResponse(response, handlers);}void SendResponse(std::shared_ptr<std::string> responseHttpFrame, std::unordered_map<std::string, std::function<void(std::string&, bool&)>>& handlers){StartTimer(timeoutSeconds);asio::async_write(*sptrSocket_, asio::buffer(*responseHttpFrame, responseHttpFrame->size()), boost::asio::bind_executor(*sptrStrand_,[this,self = shared_from_this(), responseHttpFrame, &handlers](boost::system::error_code errCode, std::size_t sendLength) {CancelTimer();if (errCode){// 常见“客户端已断开”错误码，一律静默返回//“远程主机强迫关闭了一个现有的连接”是对端（浏览器 / 测试工具）在收到响应前就把 TCP 连接 RST 掉的典型错误//在 HTTP 服务器里，这类“客户端提前断开”是正常现象，不应该当成错误打印if (errCode == boost::asio::error::connection_aborted || // WindowserrCode == boost::asio::error::eof || // 对方发送 FINerrCode == boost::asio::error::connection_reset) // RSTreturn;LogErr() << "[SendResponse] error: " << errCode.message();return;}//若keepAlive_为false，则发送一次response后立即关闭会话if (!keepAlive_){CloseSession();return;}//继续接受client的requestStartTimer(timeoutSeconds);boost::asio::async_read_until(*sptrSocket_, *(this->sptrBuffer_), "\r\n\r\n", boost::asio::bind_executor(*sptrStrand_, [this, self, &handlers](boost::system::error_code errCode, std::size_t headerLength) {////注册读取请求头//this->ParseRequesetHeader(errCode, headerLength);//再次读取、处理请求if (!errCode){CancelTimer();ParseRequest(handlers);}else{CloseSession();}}));}));}/*void SendDefaultResponse(std::string defaultBody, std::unordered_map<std::string, std::function<void(std::string&, bool&)>>& handlers){std::ostringstream defaultHttpResponseFrame;defaultHttpResponseFrame << "HTTP/1.1 404 Not Found\r\n"<< "Content-Length: " << defaultBody.size() << "\r\n"<< "Content-Type: text/plain\r\n"<< "Connection: close\r\n"<< "\r\n"<< defaultBody;auto defaultResponse = std::make_shared<std::string>(defaultHttpResponseFrame.str());SendResponse(defaultResponse, handlers);}*/void SendErrorResponse(int statusCode, const std::string& message, std::unordered_map<std::string, std::function<void(std::string&, bool&)>>& handlers){std::ostringstream response;// 简洁的纯文本错误响应std::string textBody = "Error " + std::to_string(statusCode) + ": " + message +"\nTimestamp: " + HttpServer::getCurrentTimeFormatted() +"\n";response << "HTTP/1.1 " << statusCode << " " << message << "\r\n";response << "Content-Type: text/plain; charset=utf-8\r\n";response << "Content-Length: " << textBody.length() << "\r\n";// 错误响应后通常关闭连接，避免连接状态不一致response << "Connection: close\r\n";keepAlive_ = false;response << "\r\n"; // 空行分隔头部和主体response << textBody;auto errorResponse = std::make_shared<std::string>(response.str());LogInfo() << "[Sending error response] " << statusCode << " " << message;SendResponse(errorResponse,handlers);}void CloseSession(){//socket_.shutdown(ip::tcp::socket::shutdown_both);//socket_.close();system::error_code err;sptrSocket_->shutdown(err);if (err && err != asio::error::eof && err != asio::error::connection_reset){LogErr() << "[CloseSession] shutdown err.." << err.message();}}void StartTimer(int secs){timer_.expires_after(std::chrono::seconds(secs));timer_.async_wait([self=shared_from_this()](const boost::system::error_code errCode) {// 正常 cancel，不算超时if (errCode == boost::asio::error::operation_aborted){//LogInfo() << "timer cancel...";return; }//其他异常if (errCode){LogErr() << "[StartTimer] error: " << errCode.message();return;}//超时LogInfo() << "[timeout] socket will close..";self->CloseSession();});}void CancelTimer(){//timer_.cancel();}//ip::tcp::socket socket_;//httpstd::shared_ptr<ssl::stream<ip::tcp::socket>> sptrSocket_;//httpsstd::shared_ptr<asio::streambuf> sptrBuffer_;bool keepAlive_ = true; //是否保持http长连接（http1.1默认为true，一般采取计时未通信or通信一定次数后断开）std::shared_ptr<boost::asio::strand<boost::asio::io_context::executor_type>> sptrStrand_;std::string url_; //http路径std::unordered_map<std::string, std::string> headers_; //报文头std::string body_; //报文体std::string currentHeaderField_; //当前header字段bool isParseComplete_ = false; //llhttp是否对一次请求解析完毕llhttp_t parser_;static inline llhttp_settings_t settings_;boost::asio::steady_timer timer_;void Reset(){llhttp_reset(&parser_);parser_.data = this;url_.clear();headers_.clear();body_.clear();isParseComplete_ = false;}};
};
std::mutex HttpServer::LogInfo::mtx_;
std::mutex HttpServer::LogErr::mtx_;int main()
{try{asio::io_context ioc;uint16_t port = 8080;auto server = std::make_shared<HttpServer>(ioc, port);server->RegisterHandler("/hello", [](std::string& response, bool& keep_alive) {std::string respBody = "你好世界！！";std::ostringstream oss;oss << "HTTP/1.1 200 OK\r\n"<< "Content-Length: " << respBody.size() << "\r\n"<< "Content-Type: text/plain\r\n"<< "Connection: keep-alive\r\n"<< "\r\n"<< respBody;response = oss.str();keep_alive = true;});HttpServer::LogInfo() << "[1] server start on port " << port <<"..";server->Start(); HttpServer::LogInfo() << "[2] ioc.run()..";//ioc.run(); // 阻塞在这里，调用任务链//线程池执行std::vector<std::thread> threadPool;constexpr double io_bound_factor = 0.5; // I/O密集型因子unsigned int thCount = std::max<unsigned int>(1u, io_bound_factor * std::thread::hardware_concurrency());//std::max处理线程数为0的边界条件for (unsigned int i = 0;i < thCount;++i){threadPool.emplace_back([&ioc]() {try{ioc.run();}catch (const std::exception& e){//HttpServer::LogErr() << e.what();throw;}});}for (auto& th : threadPool){th.join();}HttpServer::LogInfo() << "[3] ioc.run() returned (only after shutdown)";}catch (const std::exception& e){HttpServer::LogErr() << "exception: " << e.what();}std::cout << "Press Enter to exit..";std::cin.get();return 0;
}
