diff options
| -rw-r--r-- | json-rpc-shell.c | 288 |
1 files changed, 213 insertions, 75 deletions
diff --git a/json-rpc-shell.c b/json-rpc-shell.c index 1ff3120..a231d0b 100644 --- a/json-rpc-shell.c +++ b/json-rpc-shell.c @@ -74,6 +74,13 @@ static struct config_item g_config_table[] = // --- Main program ------------------------------------------------------------ +// HTTP/S and WS/S require significantly different handling. While for HTTP we +// can just use the cURL easy interface, with WebSockets it gets a bit more +// complicated and we implement it all by ourselves. +// +// Luckily on a higher level the application doesn't need to bother itself with +// the details and the backend API can be very simple.. + struct app_context; struct backend_iface @@ -108,18 +115,32 @@ enum ws_handler_state struct ws_context { + // Configuration: + char *endpoint; ///< Endpoint URL struct http_parser_url url; ///< Parsed URL + struct str_vector extra_headers; ///< Extra headers for the handshake - enum ws_handler_state state; ///< State - char *key; ///< Key for the current handshake - struct str *response_buffer; ///< Buffer for the response + // Events: + + bool waiting_for_event; ///< Running a separate loop to wait? + struct error *e; ///< Error while waiting for event + + ev_timer timeout_watcher; ///< Connection timeout watcher + struct str *response_buffer; ///< Buffer for the incoming messages + + // The TCP transport: int server_fd; ///< Socket FD of the server ev_io read_watcher; ///< Server FD read watcher SSL_CTX *ssl_ctx; ///< SSL context SSL *ssl; ///< SSL connection + // WebSockets protocol handling: + + enum ws_handler_state state; ///< State + char *key; ///< Key for the current handshake + http_parser hp; ///< HTTP parser bool parsing_header_value; ///< Parsing header value or field? struct str field; ///< Field part buffer @@ -131,14 +152,15 @@ struct ws_context enum ws_opcode message_opcode; ///< Opcode for the current message struct str message_data; ///< Concatenated message data - - struct str_vector extra_headers; ///< Extra headers for the handshake }; static void ws_context_init (struct ws_context *self) { memset (self, 0, sizeof *self); + ev_timer_init (&self->timeout_watcher, NULL, 0, 0); + self->server_fd = -1; + ev_io_init (&self->read_watcher, NULL, 0, 0); http_parser_init (&self->hp, HTTP_RESPONSE); str_init (&self->field); str_init (&self->value); @@ -755,14 +777,15 @@ backend_ws_on_data (struct app_context *ctx, const void *data, size_t len) struct error *e = NULL; if (!backend_ws_finish_handshake (ctx, &e)) { - print_debug ("WS handshake failed: %s", e->message); + print_error ("WS handshake failed: %s", e->message); error_free (e); return false; } + // Finished the handshake, return to caller + // (we run a separate loop to wait for the handshake to finish) self->state = WS_HANDLER_OPEN; - - // TODO: set the event loop to quit? + ev_unloop (EV_DEFAULT_ EVUNLOOP_ONE); if ((len -= n_parsed)) return ws_parser_push (&self->parser, @@ -775,9 +798,9 @@ backend_ws_on_data (struct app_context *ctx, const void *data, size_t len) if (n_parsed != len || err != HPE_OK) { if (err == HPE_CB_headers_complete) - print_debug ("WS handshake failed: %s", "missing `Upgrade' field"); + print_error ("WS handshake failed: %s", "missing `Upgrade' field"); else - print_debug ("WS handshake failed: %s", + print_error ("WS handshake failed: %s", http_errno_description (err)); return false; } @@ -785,6 +808,38 @@ backend_ws_on_data (struct app_context *ctx, const void *data, size_t len) } static void +backend_ws_close_connection (struct app_context *ctx) +{ + struct ws_context *self = &ctx->ws; + if (self->server_fd == -1) + return; + + ev_io_stop (EV_DEFAULT_ &self->read_watcher); + + if (self->ssl) + { + (void) SSL_shutdown (self->ssl); + SSL_free (self->ssl); + self->ssl = NULL; + } + + xclose (self->server_fd); + self->server_fd = -1; + + self->state = WS_HANDLER_CLOSED; + + // That would have no way of succeeding + // XXX: what if we're waiting for the close? + if (self->waiting_for_event) + { + if (!self->e) + error_set (&self->e, "unexpected connection close"); + + ev_unloop (EV_DEFAULT_ EVUNLOOP_ONE); + } +} + +static void backend_ws_on_fd_ready (EV_P_ ev_io *handle, int revents) { (void) loop; @@ -793,44 +848,46 @@ backend_ws_on_fd_ready (EV_P_ ev_io *handle, int revents) struct app_context *ctx = handle->data; struct ws_context *self = &ctx->ws; - (void) set_blocking (self->server_fd, false); - enum ws_read_result (*fill_buffer)(struct app_context *, void *, size_t *) = self->ssl ? backend_ws_fill_read_buffer_tls : backend_ws_fill_read_buffer; - bool disconnected = false; + bool close_connection = false; uint8_t buf[8192]; while (true) { + // Try to read some data in a non-blocking manner size_t n_read = sizeof buf; - switch (fill_buffer (ctx, buf, &n_read)) + (void) set_blocking (self->server_fd, false); + enum ws_read_result result = fill_buffer (ctx, buf, &n_read); + (void) set_blocking (self->server_fd, true); + + switch (result) { case WS_READ_AGAIN: goto end; case WS_READ_ERROR: print_error ("reading from the server failed"); - disconnected = true; + close_connection = true; goto end; case WS_READ_EOF: print_status ("the server closed the connection"); - disconnected = true; + close_connection = true; goto end; case WS_READ_OK: - // XXX: this is a bit ugly - (void) set_blocking (self->server_fd, true); - // TODO: use the return value - backend_ws_on_data (ctx, buf, n_read); - (void) set_blocking (self->server_fd, false); - break; + if (backend_ws_on_data (ctx, buf, n_read)) + break; + + // XXX: maybe we should wait until we receive an EOF + close_connection = true; + goto end; } } end: - (void) set_blocking (self->server_fd, true); - if (disconnected) - ; // TODO + if (close_connection) + backend_ws_close_connection (ctx); } static bool @@ -855,8 +912,6 @@ backend_ws_write (struct app_context *ctx, const void *data, size_t len) print_debug ("%s: %s: %s", __func__, "write", strerror (errno)); result = false; } - - // TODO: destroy the connection on failure? return result; } @@ -1012,7 +1067,7 @@ backend_ws_send_message (struct app_context *ctx, return result; } -static void +static bool backend_ws_send_control (struct app_context *ctx, enum ws_opcode opcode, const void *data, size_t len) { @@ -1023,22 +1078,22 @@ backend_ws_send_control (struct app_context *ctx, len = WS_MAX_CONTROL_PAYLOAD_LEN; } - backend_ws_send_control (ctx, opcode, data, len); + return backend_ws_send_message (ctx, opcode, data, len); } -static void +static bool backend_ws_fail (struct app_context *ctx, enum ws_status reason) { + struct ws_context *self = &ctx->ws; + uint8_t payload[2] = { reason << 8, reason }; - backend_ws_send_control (ctx, WS_OPCODE_CLOSE, payload, sizeof payload); - - // TODO: set the close timer, ignore all further incoming input (either set - // some flag for the case that we're in the middle of backend_ws_push(), - // and/or add a mechanism to stop the caller from polling the socket for - // reads). - // TODO: set the state to FAILED (not CLOSED as that means the TCP - // connection is closed) and wait until all is sent? - // TODO: make sure we don't send pings after the close + (void) backend_ws_send_control (ctx, WS_OPCODE_CLOSE, + payload, sizeof payload); + + // The caller should immediately proceed to close the TCP connection, + // e.g. by returning false from a handler + self->state = WS_HANDLER_CLOSING; + return false; } static bool @@ -1057,37 +1112,66 @@ backend_ws_on_frame_header (void *user_data, const struct ws_parser *parser) || (!ws_is_control_frame (parser->opcode) && (self->expecting_continuation && parser->opcode != WS_OPCODE_CONT)) || parser->payload_len >= 0x8000000000000000ULL) - backend_ws_fail (ctx, WS_STATUS_PROTOCOL_ERROR); + return backend_ws_fail (ctx, WS_STATUS_PROTOCOL_ERROR); else if (parser->payload_len > BACKEND_WS_MAX_PAYLOAD_LEN) - backend_ws_fail (ctx, WS_STATUS_MESSAGE_TOO_BIG); - else - return true; - return false; + return backend_ws_fail (ctx, WS_STATUS_MESSAGE_TOO_BIG); + return true; +} + +static bool +backend_ws_finish_closing_handshake + (struct app_context *ctx, const struct ws_parser *parser) +{ + struct str reason; + str_init (&reason); + + if (parser->payload_len >= 2) + { + struct msg_unpacker unpacker; + msg_unpacker_init (&unpacker, parser->input.str, parser->payload_len); + + uint16_t status_code; + msg_unpacker_u16 (&unpacker, &status_code); + print_debug ("close status code: %d", status_code); + + str_append_data (&reason, + parser->input.str + 2, parser->payload_len - 2); + } + + char *s = iconv_xstrdup (ctx->term_from_utf8, + reason.str, reason.len, NULL); + print_status ("server closed the connection (%s)", s); + str_free (&reason); + free (s); + + return backend_ws_send_control (ctx, WS_OPCODE_CLOSE, + parser->input.str, parser->payload_len); } static bool backend_ws_on_control_frame (struct app_context *ctx, const struct ws_parser *parser) { + struct ws_context *self = &ctx->ws; switch (parser->opcode) { case WS_OPCODE_CLOSE: - // TODO: confirm the close - // TODO: change the state to CLOSING - // TODO: call "on_close" - // NOTE: the reason is an empty string if omitted - break; + // We've received an unsolicited server close + if (self->state != WS_HANDLER_CLOSING) + (void) backend_ws_finish_closing_handshake (ctx, parser); + + return false; case WS_OPCODE_PING: - backend_ws_send_control (ctx, WS_OPCODE_PONG, - parser->input.str, parser->payload_len); + if (!backend_ws_send_control (ctx, WS_OPCODE_PONG, + parser->input.str, parser->payload_len)) + return false; break; case WS_OPCODE_PONG: // Not sending any pings but w/e break; default: // Unknown control frame - backend_ws_fail (ctx, WS_STATUS_PROTOCOL_ERROR); - return false; + return backend_ws_fail (ctx, WS_STATUS_PROTOCOL_ERROR); } return true; } @@ -1099,19 +1183,16 @@ backend_ws_on_message (struct app_context *ctx, struct ws_context *self = &ctx->ws; if (type != WS_OPCODE_TEXT) - { - backend_ws_fail (ctx, WS_STATUS_UNSUPPORTED_DATA); - return false; - } + return backend_ws_fail (ctx, WS_STATUS_UNSUPPORTED_DATA); - if (!self->response_buffer) + if (!self->waiting_for_event || !self->response_buffer) { - // TODO: warn about unexpected messages + print_warning ("unexpected message received"); return true; } str_append_data (self->response_buffer, data, len); - // TODO: exit the event loop + ev_unloop (EV_DEFAULT_ EVUNLOOP_ONE); return true; } @@ -1126,10 +1207,7 @@ backend_ws_on_frame (void *user_data, const struct ws_parser *parser) // TODO: do this rather in "on_frame_header" if (self->message_data.len + parser->payload_len > BACKEND_WS_MAX_PAYLOAD_LEN) - { - backend_ws_fail (ctx, WS_STATUS_MESSAGE_TOO_BIG); - return false; - } + return backend_ws_fail (ctx, WS_STATUS_MESSAGE_TOO_BIG); if (!self->expecting_continuation) self->message_opcode = parser->opcode; @@ -1143,10 +1221,7 @@ backend_ws_on_frame (void *user_data, const struct ws_parser *parser) if (self->message_opcode == WS_OPCODE_TEXT && !utf8_validate (self->parser.input.str, self->parser.input.len)) - { - backend_ws_fail (ctx, WS_STATUS_INVALID_PAYLOAD_DATA); - return false; - } + return backend_ws_fail (ctx, WS_STATUS_INVALID_PAYLOAD_DATA); bool result = backend_ws_on_message (ctx, self->message_opcode, self->message_data.str, self->message_data.len); @@ -1154,6 +1229,20 @@ backend_ws_on_frame (void *user_data, const struct ws_parser *parser) return result; } +static void +backend_ws_on_connection_timeout (EV_P_ ev_io *handle, int revents) +{ + (void) loop; + (void) revents; + + struct app_context *ctx = handle->data; + struct ws_context *self = &ctx->ws; + + hard_assert (self->waiting_for_event); + error_set (&self->e, "connection timeout"); + backend_ws_close_connection (ctx); +} + static bool backend_ws_connect (struct app_context *ctx, struct error **e) { @@ -1223,7 +1312,7 @@ backend_ws_connect (struct app_context *ctx, struct error **e) http_parser_init (&self->hp, HTTP_RESPONSE); str_reset (&self->field); str_reset (&self->value); - // TODO: write a function to free self->headers and call it here + str_map_clear (&self->headers); ws_parser_free (&self->parser); ws_parser_init (&self->parser); self->parser.on_frame_header = backend_ws_on_frame_header; @@ -1235,10 +1324,28 @@ backend_ws_connect (struct app_context *ctx, struct error **e) self->read_watcher.data = ctx; ev_io_start (EV_DEFAULT_ &self->read_watcher); - // TODO: set a timeout timer - // TODO: run an event loop to process the handshake response + // XXX: we should do everything non-blocking and include establishing + // the TCP connection in the timeout, but that requires a rewrite. + // As it is, this isn't really too useful. + ev_timer_init (&self->timeout_watcher, + backend_ws_on_connection_timeout, 30, 0); + + // Run an event loop to process the handshake + ev_timer_start (EV_DEFAULT_ &self->timeout_watcher); + self->waiting_for_event = true; + + ev_loop (EV_DEFAULT_ 0); + + self->waiting_for_event = false; + ev_timer_stop (EV_DEFAULT_ &self->timeout_watcher); - result = true; + if (self->e) + { + error_propagate (e, self->e); + self->e = NULL; + } + else + result = true; fail_2: if (!result) @@ -1269,16 +1376,28 @@ backend_ws_make_call (struct app_context *ctx, if (backend_ws_send_message (ctx, WS_OPCODE_TEXT, request, strlen (request))) break; - print_status ("connection failed"); + print_status ("connection failed, reconnecting"); if (!backend_ws_connect (ctx, e)) return false; } if (expect_content) { + // Run an event loop to retrieve the response self->response_buffer = buf; - // TODO: run an event loop to retrieve the answer into "buf" + self->waiting_for_event = true; + + ev_loop (EV_DEFAULT_ 0); + + self->waiting_for_event = false; self->response_buffer = NULL; + + if (self->e) + { + error_propagate (e, self->e); + self->e = NULL; + return false; + } } return true; } @@ -1286,7 +1405,26 @@ backend_ws_make_call (struct app_context *ctx, static void backend_ws_destroy (struct app_context *ctx) { - // TODO + struct ws_context *self = &ctx->ws; + + // TODO: maybe attempt a graceful shutdown, but for that there should + // probably be another backend method that runs an event loop + if (self->server_fd != -1) + backend_ws_close_connection (ctx); + + free (self->endpoint); + str_vector_free (&self->extra_headers); + if (self->e) + error_free (self->e); + ev_timer_stop (EV_DEFAULT_ &self->timeout_watcher); + if (self->ssl_ctx) + SSL_CTX_free (self->ssl_ctx); + free (self->key); + str_free (&self->field); + str_free (&self->value); + str_map_free (&self->headers); + ws_parser_free (&self->parser); + str_free (&self->message_data); } static struct backend_iface g_backend_ws = |