thebestvibedev/tromcho.net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement HTTP server support O_o

Browse Source
This commit is contained in:
Vitaliy Filippov
2026-02-19 12:38:16 +03:00
parent 79f9631f03
commit 8d93a97374
3 changed files with 191 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/etcd_state_client.cpp
+4 -4
View File
@@ -82,7 +82,7 @@ http_context_t *etcd_state_client_t::get_http_ctx()
if (!http_ctx)
{
std::string error;
http_ctx = http_context_init(etcd_client_cert, etcd_client_key, etcd_ca, error);
http_ctx = http_context_init(etcd_client_cert, etcd_client_key, etcd_ca, true, error);
if (!http_ctx)
{
fprintf(stderr, "Failed to initialize HTTP context: %s\n", error.c_str());
@@ -112,7 +112,7 @@ void etcd_state_client_t::etcd_call_oneshot(std::string etcd_address, std::strin
"Connection: close\r\n"
"\r\n"+req;
auto http_cli = http_init(tfd, get_http_ctx());
auto cb = [http_cli, callback](const http_message_t *response)
auto cb = [http_cli, callback](http_message_t *response)
{
std::string err;
json11::Json data;
@@ -152,7 +152,7 @@ void etcd_state_client_t::etcd_call(std::string api, json11::Json payload, int t
"\r\n"+req;
retries--;
auto cb = [this, api, payload, timeout, retries, interval, callback,
cur_addr = selected_etcd_address](const http_message_t *response)
cur_addr = selected_etcd_address](http_message_t *response)
{
std::string err;
json11::Json data;
@@ -378,7 +378,7 @@ void etcd_state_client_t::start_etcd_watcher()
else
http_close(etcd_watch_ws);
open_websocket(etcd_watch_ws, etcd_address, etcd_api_path+"/watch", { .timeout = etcd_slow_timeout, .ssl = ssl },
[this, cur_addr = selected_etcd_address](const http_message_t *msg)
[this, cur_addr = selected_etcd_address](http_message_t *msg)
{
if (msg->body.length())
{
src/client/http_client.cpp
+176 -46
View File
@@ -32,7 +32,7 @@
static std::string ws_format_frame(int type, uint64_t size);
static bool ws_parse_frame(std::string & buf, uint8_t & type, std::string & res);
static void parse_http_headers(std::string & res, http_message_t *parsed);
static void parse_http_headers(std::string & res, http_message_t *parsed, bool is_request);
struct http_context_t
{
@@ -56,6 +56,14 @@ struct http_context_t
}
};
struct http_call_t
{
std::string host;
std::string request;
http_options_t options;
std::function<void(http_message_t *)> cb;
};
struct http_co_t
{
http_context_t *ctx = NULL;
@@ -65,7 +73,7 @@ struct http_co_t
#endif
timerfd_manager_t *tfd;
std::function<void(const http_message_t*)> response_callback;
std::function<void(http_message_t*)> response_callback;
int request_timeout = 0;
bool ssl = false;
@@ -76,7 +84,7 @@ struct http_co_t
bool want_streaming;
bool keepalive;
std::vector<std::function<void()>> keepalive_queue;
std::vector<http_call_t> keepalive_queue;
int state = 0;
std::string connected_host;
@@ -112,8 +120,9 @@ struct http_co_t
void on_ssl_error(int res);
#endif
void post_message(uint8_t type, const std::string & msg);
void reply(const std::string & msg);
void send_request(const std::string & host, const std::string & request,
const http_options_t & options, std::function<void(const http_message_t *response)> response_callback);
const http_options_t & options, std::function<void(http_message_t *response)> response_callback);
};
#define HTTP_CO_CLOSED 0
@@ -124,10 +133,14 @@ struct http_co_t
#define HTTP_CO_WEBSOCKET 5
#define HTTP_CO_CHUNKED 6
#define HTTP_CO_KEEPALIVE 7
#define HTTP_CO_SERVER 8
#define HTTP_CO_REQ_HDR_RECEIVED 9
#define HTTP_CO_REQUEST_RECEIVED 10
#define DEFAULT_TIMEOUT 5000
http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key, const std::string & ssl_ca, std::string & error)
http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key,
const std::string & ssl_ca, bool verify_peer, std::string & error)
{
http_context_t *ctx = new http_context_t;
#ifdef WITH_OPENSSL
@@ -138,7 +151,7 @@ http_context_t* http_context_init(const std::string & ssl_cert, const std::strin
ctx->ssl_ctx = ssl_ctx;
if (!ssl_ctx)
goto init_err;
SSL_CTX_set_verify(ssl_ctx, SSL_VERIFY_PEER, NULL);
SSL_CTX_set_verify(ssl_ctx, verify_peer ? SSL_VERIFY_PEER : SSL_VERIFY_NONE, NULL);
if (!SSL_CTX_set_min_proto_version(ssl_ctx, TLS1_2_VERSION))
goto init_err;
if ((ssl_ca != "")
@@ -172,7 +185,7 @@ http_co_t *http_init(timerfd_manager_t *tfd, http_context_t *ctx)
}
void open_websocket(http_co_t *handler, const std::string & host, const std::string & path,
const http_options_t & options, std::function<void(const http_message_t *msg)> response_callback)
const http_options_t & options, std::function<void(http_message_t *msg)> response_callback)
{
if (handler->state == HTTP_CO_KEEPALIVE && (handler->connected_host != host || handler->ssl != options.ssl))
handler->close_connection();
@@ -200,15 +213,39 @@ void open_websocket(http_co_t *handler, const std::string & host, const std::str
}
void http_request(http_co_t *handler, const std::string & host, const std::string & request,
const http_options_t & options, std::function<void(const http_message_t *response)> response_callback)
const http_options_t & options, std::function<void(http_message_t *response)> response_callback)
{
handler->send_request(host, request, options, response_callback);
}
void http_serve(http_co_t *handler, int peer_fd, const http_options_t & options, std::function<void(http_message_t *msg)> request_callback)
{
if (handler->state != HTTP_CO_SERVER || handler->peer_fd != peer_fd)
handler->close_connection();
handler->host = "";
handler->request_timeout = options.timeout < 0 ? -1 : (options.timeout == 0 ? DEFAULT_TIMEOUT : options.timeout);
handler->want_streaming = false;
handler->keepalive = false;
handler->ssl = options.ssl;
handler->request = "";
handler->response_callback = request_callback;
handler->ws_outbox = "";
handler->response = "";
handler->sent = 0;
handler->parsed = {};
handler->peer_fd = peer_fd;
handler->state = HTTP_CO_SERVER;
handler->tfd->set_fd_handler(peer_fd, false, [handler](int peer_fd, int epoll_events)
{
handler->epoll_events |= epoll_events;
handler->handle_events();
});
}
void http_co_t::run_cb_and_clear()
{
parsed.eof = true;
std::function<void(const http_message_t*)> cb;
std::function<void(http_message_t*)> cb;
cb.swap(response_callback);
// Call callback after clearing it because otherwise we may hit reenterability problems
if (cb != NULL)
@@ -217,7 +254,7 @@ void http_co_t::run_cb_and_clear()
}
void http_co_t::send_request(const std::string & host, const std::string & request,
const http_options_t & options, std::function<void(const http_message_t *response)> response_callback)
const http_options_t & options, std::function<void(http_message_t *response)> response_callback)
{
stackin();
if (state == HTTP_CO_WEBSOCKET)
@@ -227,10 +264,7 @@ void http_co_t::send_request(const std::string & host, const std::string & reque
}
else if (state != HTTP_CO_KEEPALIVE && state != HTTP_CO_CLOSED)
{
keepalive_queue.push_back([this, host, request, options, response_callback]()
{
this->send_request(host, request, options, response_callback);
});
keepalive_queue.emplace_back((http_call_t){ host, request, options, std::move(response_callback) });
stackout();
return;
}
@@ -273,7 +307,7 @@ void http_co_t::send_request(const std::string & host, const std::string & reque
else
{
close_connection();
parsed = { .error = "HTTP request timed out" };
parsed = { .error = "HTTP request timed out", .status_code = ETIMEDOUT };
run_cb_and_clear();
}
stackout();
@@ -287,6 +321,11 @@ void http_post_message(http_co_t *handler, uint8_t type, const std::string & msg
handler->post_message(type, msg);
}
void http_reply(http_co_t *handler, const std::string & reply)
{
handler->reply(reply);
}
void http_co_t::post_message(uint8_t type, const std::string & msg)
{
stackin();
@@ -296,7 +335,8 @@ void http_co_t::post_message(uint8_t type, const std::string & msg)
request += msg;
submit_send();
}
else if (state == HTTP_CO_KEEPALIVE || state == HTTP_CO_CHUNKED)
else if (state == HTTP_CO_KEEPALIVE || state == HTTP_CO_CHUNKED ||
state == HTTP_CO_SERVER || state == HTTP_CO_REQ_HDR_RECEIVED || state == HTTP_CO_REQUEST_RECEIVED)
{
throw std::runtime_error("Attempt to send websocket message on a regular HTTP connection");
}
@@ -308,6 +348,18 @@ void http_co_t::post_message(uint8_t type, const std::string & msg)
stackout();
}
void http_co_t::reply(const std::string & reply)
{
stackin();
if (state != HTTP_CO_REQUEST_RECEIVED)
{
throw std::runtime_error("Attempt to send HTTP response in invalid connection state");
}
request += reply;
submit_send();
stackout();
}
void http_destroy(http_co_t *handler)
{
handler->end();
@@ -392,7 +444,7 @@ void http_co_t::start_ws_connection()
if (state != HTTP_CO_WEBSOCKET)
{
close_connection();
parsed = { .error = "Websocket connection timed out" };
parsed = { .error = "Websocket connection timed out", .status_code = ETIMEDOUT };
run_cb_and_clear();
}
stackout();
@@ -408,7 +460,7 @@ void http_co_t::start_connection()
if (!string_to_addr(host.c_str(), 1, 80, &addr))
{
close_connection();
parsed = { .error = "Invalid address: "+host };
parsed = { .error = "Invalid address: "+host, .status_code = EINVAL };
run_cb_and_clear();
stackout();
return;
@@ -417,7 +469,7 @@ void http_co_t::start_connection()
if (peer_fd < 0)
{
close_connection();
parsed = { .error = std::string("socket: ")+strerror(errno) };
parsed = { .error = std::string("socket: ")+strerror(errno), .status_code = errno };
run_cb_and_clear();
stackout();
return;
@@ -466,7 +518,7 @@ init_err:
if (r < 0 && errno != EINPROGRESS)
{
close_connection();
parsed = { .error = std::string("connect: ")+strerror(errno) };
parsed = { .error = std::string("connect: ")+strerror(errno), .status_code = errno };
run_cb_and_clear();
stackout();
return;
@@ -501,6 +553,8 @@ void http_co_t::handle_events()
{
if (state == HTTP_CO_HEADERS_RECEIVED)
std::swap(parsed.body, response);
else if (state == HTTP_CO_SERVER || state == HTTP_CO_REQ_HDR_RECEIVED || state == HTTP_CO_REQUEST_RECEIVED)
parsed = { .error = "client has disconnected normally" };
close_connection();
run_cb_and_clear();
break;
@@ -522,7 +576,7 @@ void http_co_t::handle_connect_result()
if (result != 0)
{
close_connection();
parsed = { .error = std::string("connect: ")+strerror(result) };
parsed = { .error = std::string("connect: ")+strerror(result), .status_code = result };
run_cb_and_clear();
stackout();
return;
@@ -587,13 +641,33 @@ again:
else if (res < 0)
{
close_connection();
parsed = { .error = std::string("sendmsg: ")+strerror(errno) };
parsed = { .error = std::string("sendmsg: ")+strerror(errno), .status_code = errno };
run_cb_and_clear();
stackout();
return;
}
sent += res;
if (state == HTTP_CO_SENDING_REQUEST)
if (state == HTTP_CO_REQUEST_RECEIVED)
{
if (sent >= request.size())
{
if (!keepalive)
{
close_connection();
parsed = { .error = "connection is not keep-alive" };
run_cb_and_clear();
stackout();
return;
}
state = HTTP_CO_SERVER;
request = "";
sent = 0;
}
else
goto again;
handle_read();
}
else if (state == HTTP_CO_SENDING_REQUEST)
{
if (sent >= request.size())
state = HTTP_CO_REQUEST_SENT;
@@ -664,7 +738,7 @@ again:
{
// Timeout happened and there is no data to read
close_connection();
parsed = { .error = "HTTP request timed out" };
parsed = { .error = "HTTP request timed out", .status_code = ETIMEDOUT };
run_cb_and_clear();
}
}
@@ -679,9 +753,11 @@ again:
epoll_events = epoll_events & ~EPOLLIN;
if (state == HTTP_CO_HEADERS_RECEIVED)
std::swap(parsed.body, response);
close_connection();
if (res < 0)
parsed = { .error = std::string("recvmsg: ")+strerror(-res) };
parsed = { .error = std::string("recvmsg: ")+strerror(-res), .status_code = (int)-res };
else if (state == HTTP_CO_SERVER || state == HTTP_CO_REQ_HDR_RECEIVED || state == HTTP_CO_REQUEST_RECEIVED)
parsed = { .error = "client has disconnected normally" };
close_connection();
run_cb_and_clear();
}
else
@@ -703,7 +779,7 @@ void http_co_t::on_ssl_error(int res)
parsed = { .error = "peer closed the SSL connection" };
}
else
parsed = { .error = std::string("SSL error: ")+ERR_error_string(ERR_get_error(), NULL) };
parsed = { .error = std::string("SSL error: ")+ERR_error_string(ERR_get_error(), NULL), .status_code = EIO };
run_cb_and_clear();
}
@@ -730,7 +806,7 @@ bool http_co_t::do_ssl_handshake(bool init_send)
else
{
int errcode = ERR_get_error();
parsed = { .error = ERR_error_string(errcode, NULL) };
parsed = { .error = ERR_error_string(errcode, NULL), .status_code = EIO };
close_connection();
run_cb_and_clear();
return false;
@@ -743,7 +819,10 @@ bool http_co_t::do_ssl_handshake(bool init_send)
bool http_co_t::handle_read()
{
stackin();
if (state == HTTP_CO_REQUEST_SENT)
if (state == HTTP_CO_REQUEST_RECEIVED)
{
}
else if (state == HTTP_CO_REQUEST_SENT)
{
int pos = response.find("\r\n\r\n");
if (pos >= 0)
@@ -755,7 +834,7 @@ bool http_co_t::handle_read()
timeout_id = -1;
}
state = HTTP_CO_HEADERS_RECEIVED;
parse_http_headers(response, &parsed);
parse_http_headers(response, &parsed, false);
if (parsed.status_code == 101 &&
parsed.headers.find("sec-websocket-accept") != parsed.headers.end() &&
parsed.headers["upgrade"] == "websocket" &&
@@ -779,7 +858,7 @@ bool http_co_t::handle_read()
{
// Sorry, unsupported response
close_connection();
parsed = { .error = "Response has neither Connection: close, nor Transfer-Encoding: chunked nor Content-Length headers" };
parsed = { .error = "Response has neither Connection: close, nor Transfer-Encoding: chunked nor Content-Length headers", .status_code = EINVAL };
run_cb_and_clear();
stackout();
return false;
@@ -791,14 +870,57 @@ bool http_co_t::handle_read()
}
}
}
if (state == HTTP_CO_HEADERS_RECEIVED && target_response_size > 0 && response.size() >= target_response_size)
else if (state == HTTP_CO_SERVER)
{
int pos = response.find("\r\n\r\n");
if (pos >= 0)
{
if (timeout_id >= 0)
{
// Timeout is cleared when headers are received
tfd->clear_timer(timeout_id);
timeout_id = -1;
}
state = HTTP_CO_REQ_HDR_RECEIVED;
parse_http_headers(response, &parsed, true);
auto conn_it = parsed.headers.find("connection");
keepalive = (conn_it != parsed.headers.end() && conn_it->second == "keep-alive");
auto enc_it = parsed.headers.find("transfer-encoding");
if (enc_it != parsed.headers.end())
{
// Sorry, unsupported request
close_connection();
parsed = { .error = "Chunked requests are not supported", .status_code = EINVAL };
run_cb_and_clear();
stackout();
return false;
}
auto len_it = parsed.headers.find("content-length");
target_response_size = stoull_full(len_it != parsed.headers.end() ? len_it->second : "");
if (!target_response_size)
{
state = HTTP_CO_REQUEST_RECEIVED;
response_callback(&parsed);
}
}
}
if ((state == HTTP_CO_HEADERS_RECEIVED || state == HTTP_CO_REQ_HDR_RECEIVED) &&
target_response_size > 0 && response.size() >= target_response_size)
{
std::swap(parsed.body, response);
if (!keepalive)
close_connection();
if (state == HTTP_CO_REQ_HDR_RECEIVED)
{
state = HTTP_CO_REQUEST_RECEIVED;
response_callback(&parsed);
}
else
state = HTTP_CO_KEEPALIVE;
run_cb_and_clear();
{
if (!keepalive)
close_connection();
else
state = HTTP_CO_KEEPALIVE;
run_cb_and_clear();
}
}
else if (state == HTTP_CO_CHUNKED && response.size() > 0)
{
@@ -861,26 +983,34 @@ void http_co_t::next_request()
{
if (keepalive_queue.size() > 0)
{
auto next = keepalive_queue[0];
keepalive_queue.erase(keepalive_queue.begin(), keepalive_queue.begin()+1);
next();
auto next = std::move(keepalive_queue[0]);
keepalive_queue.erase(keepalive_queue.begin());
send_request(next.host, next.request, next.options, next.cb);
}
}
static void parse_http_headers(std::string & res, http_message_t *parsed)
static void parse_http_headers(std::string & res, http_message_t *parsed, bool is_request)
{
int pos = res.find("\r\n");
pos = pos < 0 ? res.length() : pos+2;
std::string status_line = res.substr(0, pos);
int http_version;
char *status_text = NULL;
sscanf(status_line.c_str(), "HTTP/1.%d %d %ms", &http_version, &parsed->status_code, &status_text);
if (status_text)
if (!is_request)
{
parsed->status_line = status_text;
// %ms = allocate a buffer
free(status_text);
status_text = NULL;
sscanf(status_line.c_str(), "HTTP/1.%d %d %ms", &http_version, &parsed->status_code, &status_text);
if (status_text)
{
parsed->status_line = status_text;
// %ms = allocate a buffer
free(status_text);
status_text = NULL;
}
}
else
{
// Should be GET/POST / HTTP/1.1
parsed->status_line = status_line;
}
int prev = pos;
while ((pos = res.find("\r\n", prev)) >= prev)
src/client/http_client.h
+11 -3
View File
@@ -17,6 +17,8 @@
class timerfd_manager_t;
#pragma GCC visibility push(default)
struct http_options_t
{
int timeout;
@@ -44,13 +46,19 @@ struct http_message_t
// Opened websocket or keepalive HTTP connection
struct http_co_t;
http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key, const std::string & ssl_ca, std::string & error);
http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key,
const std::string & ssl_ca, bool verify_peer, std::string & error);
void http_context_destroy(http_context_t *ctx);
http_co_t* http_init(timerfd_manager_t *tfd, http_context_t *ctx = NULL);
void open_websocket(http_co_t *handler, const std::string & host, const std::string & path,
const http_options_t & options, std::function<void(const http_message_t *msg)> on_message);
const http_options_t & options, std::function<void(http_message_t *msg)> on_message);
void http_request(http_co_t *handler, const std::string & host, const std::string & request,
const http_options_t & options, std::function<void(const http_message_t *response)> response_callback);
const http_options_t & options, std::function<void(http_message_t *response)> response_callback);
void http_post_message(http_co_t *handler, uint8_t type, const std::string & msg);
void http_serve(http_co_t *handler, int peer_fd, const http_options_t & options,
std::function<void(http_message_t *msg)> request_callback);
void http_reply(http_co_t *handler, const std::string & reply);
void http_close(http_co_t *co);
void http_destroy(http_co_t *co);
#pragma GCC visibility pop