Skip to content

Commit

Permalink
minor fixes + support a callback for pre_http_body (on expect hea…
Browse files Browse the repository at this point in the history 
…der)
  • Loading branch information
@boazsegev
boazsegev committed Jul 11, 2023
1 parent 60b8644 commit c40789f
Show file tree
Hide file tree
Showing 5 changed files with 155 additions and 73 deletions.
68 changes: 45 additions & 23 deletions examples/client.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,9 @@ static fio_protocol_s STDIN_PROTOCOL = {
.on_data = on_input,
};

/* Opens the client connection after the server starts (avoid SIGPIPE) */
FIO_SFUNC void open_client_connection(void *is_http);

/* *****************************************************************************
The main code.
***************************************************************************** */
Expand Down Expand Up @@ -79,6 +82,8 @@ int main(int argc, char const *argv[]) {
if (fio_cli_get_i("-t") > 0)
CLIENT_PROTOCOL.timeout = (uint32_t)fio_cli_get_i("-t") * 1000;

void *is_http = NULL;

/* review CLI connection address (in URL format) */
if (!fio_cli_unnamed(0) && fio_cli_get("-b"))
fio_cli_set_unnamed(0, fio_cli_get("-b"));
Expand All @@ -88,35 +93,49 @@ int main(int argc, char const *argv[]) {
FIO_ASSERT(url_len < 1024, "URL address too long");
{ /* select connection client style / protocol */
fio_url_s url = fio_url_parse(fio_cli_unnamed(0), url_len);
if ((url.scheme.len == 4 ||
(url.scheme.len == 5 && ((url.scheme.buf[4] | 0x20) == 's'))) &&
fio_buf2u32u("http") == (fio_buf2u32u(url.scheme.buf) | 0x20202020UL)) {
/* TODO! HTTP client */
FIO_ASSERT(0, "HTTP client isn't supported yet");
} else {
if ((url.scheme.len == 2 ||
(url.scheme.len == 3 && ((url.scheme.buf[2] | 0x20) == 's'))) &&
fio_buf2u16u("ws") == (fio_buf2u16u(url.scheme.buf) | 0x2020UL)) {
/* TODO! WebSocket client */
FIO_ASSERT(0, "WebSocket client isn't supported yet");
} else {
FIO_ASSERT(fio_srv_connect(fio_cli_unnamed(0),
.protocol = &CLIENT_PROTOCOL,
.on_failed = on_failed,
.timeout = (fio_cli_get_i("-w") * 1000)),
"Connection error!");
}
/* attach STDIN */
fio_srv_attach_fd(fileno(stdin), &STDIN_PROTOCOL, NULL, NULL);
if (((url.scheme.len == 4 ||
(url.scheme.len == 5 && ((url.scheme.buf[4] | 0x20) == 's'))) &&
fio_buf2u32u("http") ==
(fio_buf2u32u(url.scheme.buf) | 0x20202020UL)) ||
((url.scheme.len == 2 ||
(url.scheme.len == 3 && ((url.scheme.buf[2] | 0x20) == 's'))) &&
fio_buf2u16u("ws") == (fio_buf2u16u(url.scheme.buf) | 0x2020UL)) ||
((url.scheme.len == 3 ||
(url.scheme.len == 4 && ((url.scheme.buf[3] | 0x20) == 's'))) &&
fio_buf2u32u("sse\xFF") == (fio_buf2u32u(url.scheme.buf) |
fio_buf2u32u("\x20\x20\x20\xFF")))) {
is_http = (void *)1;
}
}
FIO_LOG_DEBUG2("listening to user input on STDIN.");
/* set connection task in master process. */
fio_state_callback_add(FIO_CALL_ON_START, open_client_connection, is_http);
/* start server, connection termination will stop it. */
fio_srv_start(0);
printf("\n\t* connection terminated.\n");
FIO_LOG_INFO("* connection terminated.\n");
return 0;
}

/* *****************************************************************************
Opening the client connection.
***************************************************************************** */

FIO_SFUNC void open_client_connection(void *is_http) {

if (is_http) {
/* TODO! HTTP / WebSocket / SSE client */
FIO_LOG_FATAL("HTTP, WebSocket and SSE clients aren't supported yet"
"\n\t\tfor URL: %s",
fio_cli_unnamed(0));
fio_srv_stop();
} else {
FIO_ASSERT(fio_srv_connect(fio_cli_unnamed(0),
.protocol = &CLIENT_PROTOCOL,
.on_failed = on_failed,
.timeout = (fio_cli_get_i("-w") * 1000)),
"Connection error!");
}
}

/* *****************************************************************************
IO callback(s)
***************************************************************************** */
Expand All @@ -125,8 +144,11 @@ IO callback(s)
FIO_SFUNC void on_attach(fio_s *io) {
fio_subscribe(.io = io, .channel = FIO_BUF_INFO1("client"));
fio_udata_set(io, (void *)1);
printf("\t* connection established.\n");
FIO_LOG_INFO("* connection established.\n");
FIO_LOG_DEBUG2("Connected client IO to pub/sub");
/* attach STDIN */
FIO_LOG_DEBUG2("listening to user input on STDIN.");
fio_srv_attach_fd(fileno(stdin), &STDIN_PROTOCOL, NULL, NULL);
}
/** Called there's incoming data (from STDIN / the client socket. */
FIO_SFUNC void on_data(fio_s *io) {
Expand Down
80 changes: 55 additions & 25 deletions fio-stl.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30570,8 +30570,8 @@ SFUNC fio_s *fio_srv_connect FIO_NOOP(fio_srv_connect_args_s args) {
if (!args.protocol)
return NULL;
if (!args.url) {
if (args.protocol->on_close)
args.protocol->on_close(args.udata);
if (args.on_failed)
args.on_failed(args.udata);
return NULL;
}
if (!args.timeout)
Expand Down Expand Up @@ -31655,6 +31655,9 @@ FIO_CONSTRUCTOR(fio___openssl_setup_default) {
.cleanup = fio___openssl_cleanup,
};
fio_tls_default_io_functions(&FIO___OPENSSL_IO_FUNCS);
#ifdef SIGPIPE
fio_signal_monitor(SIGPIPE, NULL, NULL); /* avoid OpenSSL issue... */
#endif
}

/* *****************************************************************************
Expand Down Expand Up @@ -37687,6 +37690,8 @@ HTTP Setting Defaults
HTTP Listen
***************************************************************************** */
typedef struct fio_http_settings_s {
/** Called before body uploads, when a client sends an `Expect` header. */
void (*pre_http_body)(fio_http_s *h);
/** Callback for HTTP requests (server) or responses (client). */
void (*on_http)(fio_http_s *h);
/** (optional) the callback to be performed when the HTTP service closes. */
Expand Down Expand Up @@ -37813,9 +37818,9 @@ SFUNC fio_s *fio_http_io(fio_http_s *);
fio_subscribe(.io = fio_http_io(h), __VA_ARGS__)

/** TODO: Connects to HTTP / WebSockets / SSE connections on `url`. */
SFUNC void fio_http_connect(const char *url,
fio_http_s *h,
fio_http_settings_s settings);
SFUNC fio_s *fio_http_connect(const char *url,
fio_http_s *h,
fio_http_settings_s settings);

/** Connects to HTTP / WebSockets / SSE connections on `url`. */
#define fio_http_connect(url, h, ...) \
Expand Down Expand Up @@ -37928,6 +37933,9 @@ static void fio___http_default_on_eventsource_reconnect(fio_http_s *h,
}

static void http_settings_validate(fio_http_settings_s *s, int is_client) {
if (!s->pre_http_body)
s->pre_http_body = fio___http_default_noop;

if (!s->on_http)
s->on_http = is_client ? fio___http_default_noop
: fio___http_default_on_http_request;
Expand Down Expand Up @@ -38365,9 +38373,9 @@ void fio___http_connect_on_failed(void *udata);

void fio_http_connect___(void); /* IDE Marker */
/** Connects to HTTP / WebSockets / SSE connections on `url`. */
SFUNC void fio_http_connect FIO_NOOP(const char *url,
fio_http_s *h,
fio_http_settings_s s) {
SFUNC fio_s *fio_http_connect FIO_NOOP(const char *url,
fio_http_s *h,
fio_http_settings_s s) {
http_settings_validate(&s, 1);
fio_url_s u = (fio_url_s){0};
if (url)
Expand Down Expand Up @@ -38451,12 +38459,13 @@ SFUNC void fio_http_connect FIO_NOOP(const char *url,
.capa = p->settings.max_line_len,
.log = p->settings.log,
};
fio_srv_connect(url,
.protocol = &p->state[FIO___HTTP_PROTOCOL_ACCEPT].protocol,
.on_failed = NULL,
.udata = c,
.tls = s.tls,
.timeout = s.timeout);
return fio_srv_connect(url,
.protocol =
&p->state[FIO___HTTP_PROTOCOL_ACCEPT].protocol,
.on_failed = NULL,
.udata = c,
.tls = s.tls,
.timeout = s.timeout);
}

/* *****************************************************************************
Expand All @@ -38478,6 +38487,16 @@ static void fio_http1_on_complete(void *udata) {
HTTP/1.1 Parser callbacks
***************************************************************************** */

FIO_IFUNC void fio___http_request_too_big(fio___http_connection_s *c) {
fio_http_s *h = c->h;
fio_dup(c->io);
fio_srv_suspend(c->io);
c->h = NULL;
c->suspend = 1;
fio_http_send_error_response(h, 413);
fio_http_free(h);
}

FIO_IFUNC void fio_http1_attach_handle(fio___http_connection_s *c) {
c->h = fio_http_new();
FIO_ASSERT_ALLOC(c->h);
Expand Down Expand Up @@ -38555,6 +38574,8 @@ static int fio_http1_on_header_content_length(fio_buf_info_s name,
void *udata) {
fio___http_connection_s *c = (fio___http_connection_s *)udata;
fio_http_s *h = c->h;
if (!h)
return 0;
if (content_length > c->settings->max_body_size)
goto too_big;
if (content_length)
Expand All @@ -38567,35 +38588,44 @@ static int fio_http1_on_header_content_length(fio_buf_info_s name,
#endif
return 0;
too_big:
fio_dup(c->io);
fio_srv_suspend(c->io);
c->h = NULL;
c->suspend = 1;
fio_http_send_error_response(h, 413);
fio_http_free(h);
fio___http_request_too_big(c);
return 0; /* should we disconnect (return -1), or not? */
(void)name, (void)value;
}
/** called when `Expect` arrives and may require a 100 continue response. */
static int fio_http1_on_expect(void *udata) {
fio___http_connection_s *c = (fio___http_connection_s *)udata;
if (fio_http1_expected(&c->state.http.parser) &&
fio_http1_expected(&c->state.http.parser) != FIO_HTTP1_EXPECTED_CHUNKED &&
fio_http1_expected(&c->state.http.parser) > c->settings->max_body_size)
return -1;
fio_http_s *h = c->h;
if (!h)
return 1;
fio_dup(c->io);
c->h = NULL;
c->settings->pre_http_body(h);
if (fio_http_status(h))
goto response_sent;
c->h = h;
fio_undup(c->io);
const fio_buf_info_s response =
FIO_BUF_INFO1("HTTP/1.1 100 Continue\r\n\r\n");
fio_write2(c->io, .buf = response.buf, .len = response.len, .copy = 0);
return 0; /* TODO?: improve support for `expect` headers? */
response_sent:
fio_http_free(h);
return 1;
}

/** called when a body chunk is parsed. */
static int fio_http1_on_body_chunk(fio_buf_info_s chunk, void *udata) {
fio___http_connection_s *c = (fio___http_connection_s *)udata;
if (!c->h)
return -1; /* close connection if a large payload is unstoppable */
if (chunk.len + fio_http_body_length(c->h) > c->settings->max_body_size)
return -1;
goto too_big;
fio_http_body_write(c->h, chunk.buf, chunk.len);
return 0;
too_big:
fio___http_request_too_big(c);
return 0;
}

/* *****************************************************************************
Expand Down
4 changes: 2 additions & 2 deletions fio-stl/400 server.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2475,8 +2475,8 @@ SFUNC fio_s *fio_srv_connect FIO_NOOP(fio_srv_connect_args_s args) {
if (!args.protocol)
return NULL;
if (!args.url) {
if (args.protocol->on_close)
args.protocol->on_close(args.udata);
if (args.on_failed)
args.on_failed(args.udata);
return NULL;
}
if (!args.timeout)
Expand Down
3 changes: 3 additions & 0 deletions fio-stl/402 openssl.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -513,6 +513,9 @@ FIO_CONSTRUCTOR(fio___openssl_setup_default) {
.cleanup = fio___openssl_cleanup,
};
fio_tls_default_io_functions(&FIO___OPENSSL_IO_FUNCS);
#ifdef SIGPIPE
fio_signal_monitor(SIGPIPE, NULL, NULL); /* avoid OpenSSL issue... */
#endif
}

/* *****************************************************************************
Expand Down
Loading

0 comments on commit c40789f

Please sign in to comment.