[libuv] Call "uv_write" for http chunked in "uv_work_cb" ?

[Hussein Daneshvar]

Hi
can call "uv_write" in "uv_work_cb" for http chunked packet ?
call "uv_work_queue" in http parser "on_message_complete" and call 3
"uv_write" in "uv_work_cb" method and
call "uv_close" for close "uv_tcp_t" handle in "uv_after_work_cb" method,
this is correct ?
Should call 3 "uv_write" with separated uv_buf_t ?

thanks

[Hussein Daneshvar]

Hi
1.Initialization

- Setting "http_parser_setting"
- Init TCP for listening Port 80
- Call uv_run

2.On "uv_connection_cb"

- Call uv_tcp_init for Client handle
- Call uv_accept
- Call http_parser_init
- Call uv_read_start

3.On "uv_read_cb"

- Call http_parser_execute

4.On HTTP Parser "on_message_complete"

- Call uv_queue_work

5.On uv_work_cb

- Create new buffer and call uv_write HTTP Header with
"Transfer-Encoding: chunked" field and Data Part 1 (Free Buffer in
uv_write_cb callback)
- Create new buffer and call uv_write for Data Part 2 (Free Buffer in
uv_write_cb callback)
- Create new buffer and call uv_write for Data Part 3 (Free Buffer in
uv_write_cb callback)
- Create new buffer and call uv_write for Data Part 4 (Free Buffer in
uv_write_cb callback)

6.On uv_after_work_cb

- Call uv_close for Client handle

is this cycle right ?
Call uv_close for client handle in uv_after_work_cb callback is OK ?

static uv_loop_t* _http_loop;
static uv_tcp_t _http_handle;
static http_parser_settings _http_parser_settings;


#define MAX_HTTP_HEADERS 20


/**
* Represents a single http header.
*/
typedef struct {
const char* field;
const char* value;
size_t field_length;
size_t value_length;
} HttpHeader;


/**
* Represents a http request with internal dependencies.
*
* - write request for sending the response
* - reference to tcp socket as write stream
* - instance of http_parser parser
* - string of the http url
* - string of the http method
* - amount of total header lines
* - http header array
* - body content
*/
typedef struct {
uv_tcp_t handle;
http_parser parser;
char* url;
char* method;
int header_lines;
HttpHeader headers[MAX_HTTP_HEADERS];
const char* body;
} HttpRequest, *PHttpRequest;


typedef struct {
uv_work_t work;
uv_write_t write;
HttpRequest* http_request;
bool error;
} HttpResponse, *PHttpResponse;


typedef struct {
vString buf;
uv_tcp_t* handle;
uv_write_t write;
}HttpWrite, *PHttpWrite;


void DoClose(uv_handle_t* handle) {
free((HttpRequest*)handle->data);
}


void DoAlloc(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
*buf = uv_buf_init((char*)malloc(suggested_size), suggested_size);
}


void DoRead(uv_stream_t* tcp, ssize_t nread, const uv_buf_t * buf) {
ssize_t parsed;
HttpRequest* http_request = (HttpRequest*)tcp->data;
if (nread >= 0) {
parsed = (ssize_t)http_parser_execute(&http_request->parser, &
_http_parser_settings, buf->base, nread);
if (parsed < nread) {
uv_close((uv_handle_t*)&http_request->handle, DoClose);
}
} else {
if (nread != UV_EOF) {
UVERR(nread, "read");
}
uv_close((uv_handle_t*)&http_request->handle, DoClose);
}
free(buf->base);
}


void DoAfterWrite(uv_write_t* req, int status) {
CHECK(status, "write");
if (!uv_is_closing((uv_handle_t*)req->handle)) {
HttpResponse* http_response = (HttpResponse*)req->data;
uv_close((uv_handle_t*)req->handle, DoClose);
free(http_response);
}
}


void DoConnect(uv_stream_t* server_handle, int status) {


assert((uv_tcp_t*)server_handle == &_http_handle);
CHECK(status, "Connect");


HttpRequest* http_request = (HttpRequest*)malloc(sizeof(HttpRequest));


uv_tcp_init(_http_loop, &http_request->handle);
http_request->handle.data = http_request;
http_request->parser.data = http_request;
/* accept the created http_request */
if (uv_accept(server_handle, (uv_stream_t*)&http_request->handle) == 0) {
/* initialize our http parser */
http_parser_init(&http_request->parser, HTTP_REQUEST);
/* start reading from the tcp http_request socket */
uv_read_start((uv_stream_t*)&http_request->handle, DoAlloc, DoRead);
} else {
/* we seem to have an error and quit */
uv_close((uv_handle_t*)&http_request->handle, DoClose);
}
}


int DoMessageBegin(http_parser* parser) {
HttpRequest* http_request = (HttpRequest*)parser->data;
http_request->header_lines = 0;
return 0;
}


int DoURL(http_parser* parser, const char* at, size_t length) {
HttpRequest* http_request = (HttpRequest*)parser->data;
http_request->url = (char*)malloc(length + 1);
strncpy((char*)http_request->url, at, length);
http_request->url[length] = '\0';
return 0;
}


int DoHeaderField(http_parser* parser, const char* at, size_t length) {
HttpRequest* http_request = (HttpRequest*)parser->data;
HttpHeader* header = &http_request->headers[http_request->header_lines];
header->field = (char*)malloc(length + 1);
header->field_length = length;
strncpy((char*)header->field, at, length);
return 0;
}


int DoHeaderValue(http_parser* parser, const char* at, size_t length) {
HttpRequest* http_request = (HttpRequest*)parser->data;
HttpHeader* header = &http_request->headers[http_request->header_lines];
header->value_length = length;
header->value = (char*)malloc(length + 1);
strncpy((char*)header->value, at, length);
++http_request->header_lines;
return 0;
}


int DoHeadersComplete(http_parser* parser) {
HttpRequest* http_request = (HttpRequest*)parser->data;
const char* method = http_method_str((enum http_method)parser->method);
http_request->method = (char*)malloc(sizeof(method));
strncpy(http_request->method, method, strlen(method));
return 0;
}


int DoBody(http_parser* parser, const char* at, size_t length) {
HttpRequest* http_request = (HttpRequest*)parser->data;
http_request->body = (char*)malloc(length + 1);
http_request->body = at;
return 0;
}


void DoWorkWriteAfter(uv_write_t* req, int status) {
CHECK(status, "write");
PHttpWrite http_writer = (PHttpResponse)req->data;
vStrFree(http_writer->buf);
free(http_writer);
}


void DoWork(uv_work_t* req) {
PHttpResponse http_response = (PHttpResponse)req->data;
PHttpWrite http_write;
uv_buf_t buf;


// Write Part 1
http_write = (PHttpWrite)malloc(sizeof(HttpWrite));
http_write->handle = &http_response->http_request->handle;
http_write->buf = vStrNew("");
vStrAdd(http_write->buf, "HTTP/1.1 200 OK\r\nContent-Type: text/html;
charset=UTF-8\r\nKeep-Alive: timeout=10, max=100\r\nConnection:
keep-alive\r\nTransfer-Encoding: chunked\r\n\r\n");
vStrAdd(http_write->buf, "c\r\n<h1>go1</h1>\r\n");
http_write->write.data = http_write;
buf.base = http_write->buf;
buf.len = vStrLen(http_write->buf);
uv_write(&http_write->write,
(uv_stream_t*)http_write->handle,
&buf,
1,
DoWorkWriteAfter);


// Write Part 2
http_write = (PHttpWrite)malloc(sizeof(HttpWrite));
http_write->handle = &http_response->http_request->handle;
http_write->buf = vStrNew("");
vStrAdd(http_write->buf, "c\r\n<h1>go2</h1>\r\n");
http_write->write.data = http_write;
buf.base = http_write->buf;
buf.len = vStrLen(http_write->buf);
uv_write(&http_write->write,
(uv_stream_t*)http_write->handle,
&buf,
1,
DoWorkWriteAfter);


// Write Part 3
http_write = (PHttpWrite)malloc(sizeof(HttpWrite));
http_write->handle = &http_response->http_request->handle;
http_write->buf = vStrNew("");
vStrAdd(http_write->buf, "f\r\n<h1>go2.5</h1>\r\n");
http_write->write.data = http_write;
buf.base = http_write->buf;
buf.len = vStrLen(http_write->buf);
uv_write(&http_write->write,
(uv_stream_t*)http_write->handle,
&buf,
1,
DoWorkWriteAfter);


// Write Part 4
http_write = (PHttpWrite)malloc(sizeof(HttpWrite));
http_write->handle = &http_response->http_request->handle;
http_write->buf = vStrNew("");
vStrAdd(http_write->buf, "c\r\n<h1>go3</h1>\r\n0\r\n\r\n");
http_write->write.data = http_write;
buf.base = http_write->buf;
buf.len = vStrLen(http_write->buf);
uv_write(&http_write->write,
(uv_stream_t*)http_write->handle,
&buf,
1,
DoWorkWriteAfter);
}


void DoAfterWork(uv_work_t* req, int status) {
HttpResponse* http_response = (HttpResponse*)req->data;
if (!uv_is_closing((uv_handle_t*)&http_response->http_request->handle)) {
HttpResponse* http_response = (HttpResponse*)req->data;
uv_close((uv_handle_t*)&http_response->http_request->handle, DoClose);
free(http_response);
}
}


int DoMessageComplete(http_parser* parser) {
HttpResponse* http_response = (HttpResponse*)malloc(sizeof(HttpResponse));
http_response->work.data = http_response;
http_response->error = false;
http_response->http_request = (HttpRequest*)parser->data;
int status = uv_queue_work(_http_loop, &http_response->work, DoWork,
DoAfterWork);
CHECK(status, "uv_queue_work");
assert(status == 0);
return 0;
}


void vStartHttpServer(int Port) {
int r;
struct sockaddr_in address;


_http_parser_settings.on_message_begin = DoMessageBegin;
_http_parser_settings.on_url = DoURL;
_http_parser_settings.on_header_field = DoHeaderField;
_http_parser_settings.on_header_value = DoHeaderValue;
_http_parser_settings.on_headers_complete = DoHeadersComplete;
_http_parser_settings.on_body = DoBody;
_http_parser_settings.on_message_complete = DoMessageComplete;

_http_loop = uv_default_loop();
r = uv_tcp_init(_http_loop, &_http_handle);
CHECK(r, "TCP Init");


r = uv_tcp_keepalive(&_http_handle, 1, 60);
CHECK(r, "TCP KeepAlive");


r = uv_ip4_addr("0.0.0.0", Port, &address);
CHECK(r, "IPv4 Address");


r = uv_tcp_bind(&_http_handle, (const struct sockaddr*)&address, 0);
CHECK(r, "TCP Bind");


r = uv_listen((uv_stream_t*)&_http_handle, 128, DoConnect);
CHECK(r, "Listen");


async_write = (uv_async_t*)malloc(sizeof(uv_async_t));
uv_async_init(_http_loop, async_write, DoWorkWrite);


LOGF("Listening on port %d", Port);
uv_run(_http_loop, UV_RUN_DEFAULT);
}