kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-09-02 19:16:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement internal abstraction for iostreams

Browse Source 
fopencookie/funclose is a non-standard API and should not be used
in portable software. Additionally, the way FILE's fd is used in
non-blocking mode is undefined behaviour and cannot be relied on.

This introduces internal abstraction for io streams, that allows
implementing the desired virtualization of read/write operations
with necessary timeout handling.

ASTERISK-24515 #close
ASTERISK-24517 #close

Change-Id: Id916aef418b665ced6a7489aef74908b6e376e85
This commit is contained in:
Timo Teräs
2016-06-02 22:10:06 +03:00
parent 0cc14597b2
commit 070a51bf7c
13 changed files with 1002 additions and 1103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
res/res_http_websocket.c
View File

@@ -73,8 +73,7 @@
/*! \brief Structure definition for session */
struct ast_websocket {
FILE *f; /*!< Pointer to the file instance used for writing and reading */
int fd; /*!< File descriptor for the session, only used for polling */
struct ast_iostream *stream; /*!< iostream of the connection */
struct ast_sockaddr address; /*!< Address of the remote client */
enum ast_websocket_opcode opcode; /*!< Cached opcode for multi-frame messages */
size_t payload_len; /*!< Length of the payload */
@@ -165,10 +164,11 @@ static void session_destroy_fn(void *obj)
{
struct ast_websocket *session = obj;
if (session->f) {
if (session->stream) {
ast_websocket_close(session, 0);
if (session->f) {
fclose(session->f);
if (session->stream) {
ast_iostream_close(session->stream);
session->stream = NULL;
ast_verb(2, "WebSocket connection %s '%s' closed\n", session->client ? "to" : "from",
ast_sockaddr_stringify(&session->address));
}
@@ -294,20 +294,22 @@ int AST_OPTIONAL_API_NAME(ast_websocket_close)(struct ast_websocket *session, ui
session->close_sent = 1;
ao2_lock(session);
res = ast_careful_fwrite(session->f, session->fd, frame, 4, session->timeout);
ast_iostream_set_timeout_inactivity(session->stream, session->timeout);
res = ast_iostream_write(session->stream, frame, sizeof(frame));
ast_iostream_set_timeout_disable(session->stream);
/* If an error occurred when trying to close this connection explicitly terminate it now.
* Doing so will cause the thread polling on it to wake up and terminate.
*/
if (res) {
fclose(session->f);
session->f = NULL;
if (res != sizeof(frame)) {
ast_iostream_close(session->stream);
session->stream = NULL;
ast_verb(2, "WebSocket connection %s '%s' forcefully closed due to fatal write error\n",
session->client ? "to" : "from", ast_sockaddr_stringify(&session->address));
}
ao2_unlock(session);
return res;
return res == sizeof(frame);
}
static const char *opcode_map[] = {
@@ -375,7 +377,8 @@ int AST_OPTIONAL_API_NAME(ast_websocket_write)(struct ast_websocket *session, en
return -1;
}
if (ast_careful_fwrite(session->f, session->fd, frame, frame_size, session->timeout)) {
ast_iostream_set_timeout_sequence(session->stream, ast_tvnow(), session->timeout);
if (ast_iostream_write(session->stream, frame, frame_size) != frame_size) {
ao2_unlock(session);
/* 1011 - server terminating connection due to not being able to fulfill the request */
ast_debug(1, "Closing WS with 1011 because we can't fulfill a write request\n");
@@ -383,7 +386,7 @@ int AST_OPTIONAL_API_NAME(ast_websocket_write)(struct ast_websocket *session, en
return -1;
}
fflush(session->f);
ast_iostream_set_timeout_disable(session->stream);
ao2_unlock(session);
return 0;
@@ -411,7 +414,7 @@ void AST_OPTIONAL_API_NAME(ast_websocket_unref)(struct ast_websocket *session)
int AST_OPTIONAL_API_NAME(ast_websocket_fd)(struct ast_websocket *session)
{
return session->closing ? -1 : session->fd;
return session->closing ? -1 : ast_iostream_get_fd(session->stream);
}
struct ast_sockaddr * AST_OPTIONAL_API_NAME(ast_websocket_remote_address)(struct ast_websocket *session)
@@ -426,18 +429,8 @@ int AST_OPTIONAL_API_NAME(ast_websocket_is_secure)(struct ast_websocket *session
int AST_OPTIONAL_API_NAME(ast_websocket_set_nonblock)(struct ast_websocket *session)
{
int flags;
if ((flags = fcntl(session->fd, F_GETFL)) == -1) {
return -1;
}
flags |= O_NONBLOCK;
if ((flags = fcntl(session->fd, F_SETFL, flags)) == -1) {
return -1;
}
ast_iostream_nonblock(session->stream);
ast_iostream_set_exclusive_input(session->stream, 0);
return 0;
}
@@ -490,17 +483,16 @@ static inline int ws_safe_read(struct ast_websocket *session, char *buf, int len
int sanity = 10;
ao2_lock(session);
if (!session->f) {
if (!session->stream) {
ao2_unlock(session);
errno = ECONNABORTED;
return -1;
}
for (;;) {
clearerr(session->f);
rlen = fread(rbuf, 1, xlen, session->f);
if (!rlen) {
if (feof(session->f)) {
rlen = ast_iostream_read(session->stream, rbuf, xlen);
if (rlen != xlen) {
if (rlen == 0) {
ast_log(LOG_WARNING, "Web socket closed abruptly\n");
*opcode = AST_WEBSOCKET_OPCODE_CLOSE;
session->closing = 1;
@@ -508,7 +500,7 @@ static inline int ws_safe_read(struct ast_websocket *session, char *buf, int len
return -1;
}
if (ferror(session->f) && errno != EAGAIN) {
if (rlen < 0 && errno != EAGAIN) {
ast_log(LOG_ERROR, "Error reading from web socket: %s\n", strerror(errno));
*opcode = AST_WEBSOCKET_OPCODE_CLOSE;
session->closing = 1;
@@ -529,7 +521,7 @@ static inline int ws_safe_read(struct ast_websocket *session, char *buf, int len
if (!xlen) {
break;
}
if (ast_wait_for_input(session->fd, 1000) < 0) {
if (ast_wait_for_input(ast_iostream_get_fd(session->stream), 1000) < 0) {
ast_log(LOG_ERROR, "ast_wait_for_input returned err: %s\n", strerror(errno));
*opcode = AST_WEBSOCKET_OPCODE_CLOSE;
session->closing = 1;
@@ -824,7 +816,7 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
ao2_ref(protocol_handler, -1);
return 0;
}
session->timeout = AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
session->timeout = AST_DEFAULT_WEBSOCKET_WRITE_TIMEOUT;
/* Generate the session id */
if (!ast_uuid_generate_str(session->session_id, sizeof(session->session_id))) {
@@ -854,7 +846,8 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
* Connection_.
*/
if (protocol) {
fprintf(ser->f, "HTTP/1.1 101 Switching Protocols\r\n"
ast_iostream_printf(ser->stream,
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: %s\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: %s\r\n"
@@ -863,15 +856,14 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
websocket_combine_key(key, base64, sizeof(base64)),
protocol);
} else {
fprintf(ser->f, "HTTP/1.1 101 Switching Protocols\r\n"
ast_iostream_printf(ser->stream,
"HTTP/1.1 101 Switching Protocols\r\n"
"Upgrade: %s\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: %s\r\n\r\n",
upgrade,
websocket_combine_key(key, base64, sizeof(base64)));
}
fflush(ser->f);
} else {
/* Specification defined in http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-75 or completely unknown */
@@ -883,7 +875,7 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
}
/* Enable keepalive on all sessions so the underlying user does not have to */
if (setsockopt(ser->fd, SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags))) {
if (setsockopt(ast_iostream_get_fd(ser->stream), SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags))) {
ast_log(LOG_WARNING, "WebSocket connection from '%s' could not be accepted - failed to enable keepalive\n",
ast_sockaddr_stringify(&ser->remote_address));
websocket_bad_request(ser);
@@ -895,25 +887,23 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
ast_verb(2, "WebSocket connection from '%s' for protocol '%s' accepted using version '%d'\n", ast_sockaddr_stringify(&ser->remote_address), protocol ? : "", version);
/* Populate the session with all the needed details */
session->f = ser->f;
session->fd = ser->fd;
session->stream = ser->stream;
ast_sockaddr_copy(&session->address, &ser->remote_address);
session->opcode = -1;
session->reconstruct = DEFAULT_RECONSTRUCTION_CEILING;
session->secure = ser->ssl ? 1 : 0;
session->secure = ast_iostream_get_ssl(ser->stream) ? 1 : 0;
/* Give up ownership of the socket and pass it to the protocol handler */
ast_tcptls_stream_set_exclusive_input(ser->stream_cookie, 0);
ast_iostream_set_exclusive_input(session->stream, 0);
protocol_handler->session_established(session, get_vars, headers);
ao2_ref(protocol_handler, -1);
/*
* By dropping the FILE* and fd from the session the connection
* By dropping the stream from the session the connection
* won't get closed when the HTTP server cleans up because we
* passed the connection to the protocol handler.
*/
ser->f = NULL;
ser->fd = -1;
ser->stream = NULL;
return 0;
}
@@ -1247,7 +1237,7 @@ static enum ast_websocket_result websocket_client_handshake_get_response(
int has_accept = 0;
int has_protocol = 0;
if (!fgets(buf, sizeof(buf), client->ser->f)) {
if (ast_iostream_gets(client->ser->stream, buf, sizeof(buf)) <= 0) {
ast_log(LOG_ERROR, "Unable to retrieve HTTP status line.");
return WS_BAD_STATUS;
}
@@ -1260,7 +1250,7 @@ static enum ast_websocket_result websocket_client_handshake_get_response(
/* Ignoring line folding - assuming header field values are contained
within a single line */
while (fgets(buf, sizeof(buf), client->ser->f)) {
while (ast_iostream_gets(client->ser->stream, buf, sizeof(buf)) > 0) {
char *name, *value;
int parsed = ast_http_header_parse(buf, &name, &value);
@@ -1313,19 +1303,19 @@ static enum ast_websocket_result websocket_client_handshake(
client->protocols);
}
if (fprintf(client->ser->f,
"GET /%s HTTP/1.1\r\n"
"Sec-WebSocket-Version: %d\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Host: %s\r\n"
"Sec-WebSocket-Key: %s\r\n"
"%s\r\n",
client->resource_name ? ast_str_buffer(client->resource_name) : "",
client->version,
client->host,
client->key,
protocols) < 0) {
if (ast_iostream_printf(client->ser->stream,
"GET /%s HTTP/1.1\r\n"
"Sec-WebSocket-Version: %d\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
"Host: %s\r\n"
"Sec-WebSocket-Key: %s\r\n"
"%s\r\n",
client->resource_name ? ast_str_buffer(client->resource_name) : "",
client->version,
client->host,
client->key,
protocols) < 0) {
ast_log(LOG_ERROR, "Failed to send handshake.\n");
return WS_WRITE_ERROR;
}
@@ -1349,9 +1339,9 @@ static enum ast_websocket_result websocket_client_connect(struct ast_websocket *
return res;
}
ws->f = ws->client->ser->f;
ws->fd = ws->client->ser->fd;
ws->secure = ws->client->ser->ssl ? 1 : 0;
ws->stream = ws->client->ser->stream;
ws->secure = ast_iostream_get_ssl(ws->stream) ? 1 : 0;
ws->client->ser->stream = NULL;
ast_sockaddr_copy(&ws->address, &ws->client->ser->remote_address);
return WS_OK;
}