summaryrefslogtreecommitdiff
path: root/demo-json-rpc-server.c
diff options
context:
space:
mode:
Diffstat (limited to 'demo-json-rpc-server.c')
-rw-r--r--demo-json-rpc-server.c314
1 files changed, 272 insertions, 42 deletions
diff --git a/demo-json-rpc-server.c b/demo-json-rpc-server.c
index 1b10e9d..77c9ca5 100644
--- a/demo-json-rpc-server.c
+++ b/demo-json-rpc-server.c
@@ -701,6 +701,12 @@ fcgi_request_write (struct fcgi_request *self, const void *data, size_t len)
}
}
+static void
+fcgi_request_finish (struct fcgi_request *self)
+{
+ // TODO: flush(), end_request(), delete self, muxer->request_destroy_cb()?
+}
+
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
typedef void (*fcgi_muxer_handler_fn)
@@ -764,6 +770,7 @@ fcgi_muxer_on_begin_request
return;
}
+ // We can only act as a responder, reject everything else up front
if (role != FCGI_RESPONDER)
{
fcgi_muxer_send_end_request (self,
@@ -1063,6 +1070,8 @@ scgi_parser_push (struct scgi_parser *self,
#define SEC_WS_PROTOCOL "Sec-WebSocket-Protocol"
#define SEC_WS_VERSION "Sec-WebSocket-Version"
+#define WS_MAX_CONTROL_PAYLOAD_LEN 125
+
static char *
ws_encode_response_key (const char *key)
{
@@ -1122,6 +1131,12 @@ enum ws_opcode
WS_OPCODE_PONG = 10
};
+static bool
+ws_is_control_frame (int opcode)
+{
+ return opcode >= WS_OPCODE_CLOSE;
+}
+
struct ws_parser
{
struct str input; ///< External input buffer
@@ -1136,8 +1151,7 @@ struct ws_parser
uint32_t mask; ///< Frame mask
uint64_t payload_len; ///< Payload length
- // TODO: it wouldn't be half bad if there was a callback to just validate
- // the frame header (such as the maximum payload length)
+ bool (*on_frame_header) (void *user_data, const struct ws_parser *self);
/// Callback for when a message is successfully parsed.
/// The actual payload is stored in "input", of length "payload_len".
@@ -1248,17 +1262,17 @@ ws_parser_push (struct ws_parser *self, const void *data, size_t len)
case WS_PARSER_MASK:
if (!self->is_masked)
- {
- self->state = WS_PARSER_PAYLOAD;
- break;
- }
+ goto end_of_header;
if (self->input.len < 4)
return true;
(void) msg_unpacker_u32 (&unpacker, &self->mask);
+ str_remove_slice (&self->input, 0, 4);
+ end_of_header:
self->state = WS_PARSER_PAYLOAD;
- str_remove_slice (&self->input, 0, 4);
+ if (!self->on_frame_header (self->user_data, self))
+ return false;
break;
case WS_PARSER_PAYLOAD:
@@ -1289,7 +1303,7 @@ ws_parser_push (struct ws_parser *self, const void *data, size_t len)
enum ws_handler_state
{
- WS_HANDLER_HTTP, ///< Parsing HTTP
+ WS_HANDLER_HANDSHAKE, ///< Parsing HTTP
WS_HANDLER_WEBSOCKETS ///< Parsing WebSockets frames
};
@@ -1305,12 +1319,38 @@ struct ws_handler
struct str url; ///< Request URL
struct ws_parser parser; ///< Protocol frame parser
+ bool expecting_continuation; ///< For non-control traffic
+
+ enum ws_opcode message_opcode; ///< Opcode for the current message
+ struct str message_data; ///< Concatenated message data
+
+ unsigned ping_interval; ///< Ping interval in seconds
+ uint64_t max_payload_len; ///< Maximum length of any message
+
+ // TODO: bool closing; // XXX: rather a { OPEN, CLOSING } state?
+ // TODO: a close timer
- // TODO: bool closing;
- // TODO: a configurable max_payload_len initialized by _init()
+ // TODO: a ping timer (when no pong is received by the second time the
+ // timer triggers, it is a ping timeout)
+ ev_timer ping_timer; ///< Ping timer
+ bool received_pong; ///< Received PONG since the last PING
/// Called upon reception of a single full message
- bool (*on_message) (void *user_data, const void *data, size_t len);
+ bool (*on_message) (void *user_data,
+ enum ws_opcode type, const void *data, size_t len);
+
+ // TODO: void (*on_initialized) () that will allow the user to choose
+ // any sub-protocol, if the client has provided any.
+
+ /// The connection has been closed.
+ /// @a close_code may, or may not, be one of enum ws_status.
+ // NOTE: the "close_code" is what we receive from the remote endpoint,
+ // or one of 1005/1006/1015
+ // NOTE: the reason is an empty string if omitted
+ // TODO; also note that ideally, the handler should (be able to) first
+ // receive a notification about the connection being closed because of
+ // an error (recv()) returns -1, and call on_close() in reaction.
+ void (*on_close) (void *user_data, int close_code, const char *reason);
/// Write a chunk of data to the stream
void (*write_cb) (void *user_data, const void *data, size_t len);
@@ -1320,15 +1360,126 @@ struct ws_handler
void *user_data; ///< User data for callbacks
};
+static void
+ws_handler_send_control (struct ws_handler *self, enum ws_opcode opcode,
+ const void *data, size_t len)
+{
+ if (len > WS_MAX_CONTROL_PAYLOAD_LEN)
+ {
+ print_debug ("truncating output control frame payload"
+ " from %zu to %zu bytes", len, (size_t) WS_MAX_CONTROL_PAYLOAD_LEN);
+ len = WS_MAX_CONTROL_PAYLOAD_LEN;
+ }
+
+ uint8_t header[2] = { 0x80 | (opcode & 0x0F), len };
+ self->write_cb (self->user_data, header, sizeof header);
+ self->write_cb (self->user_data, data, len);
+}
+
+static void
+ws_handler_fail (struct ws_handler *self, enum ws_status reason)
+{
+ uint8_t payload[2] = { reason << 8, reason };
+ ws_handler_send_control (self, 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 ws_handler_push(),
+ // and/or add a mechanism to stop the caller from polling the socket for
+ // reads).
+}
+
+// TODO: ws_handler_close() that behaves like ws_handler_fail() but doesn't
+// ignore frames up to a corresponding close from the client.
+// Read the RFC once again to see if we can really process the frames.
+
+static bool
+ws_handler_on_frame_header (void *user_data, const struct ws_parser *parser)
+{
+ struct ws_handler *self = user_data;
+
+ if (parser->reserved_1 || parser->reserved_2 || parser->reserved_3
+ || !parser->is_masked // client -> server payload must be masked
+ || (ws_is_control_frame (parser->opcode) &&
+ (!parser->is_fin || parser->payload_len > WS_MAX_CONTROL_PAYLOAD_LEN))
+ || (!ws_is_control_frame (parser->opcode) &&
+ (self->expecting_continuation && parser->opcode != WS_OPCODE_CONT)))
+ ws_handler_fail (self, WS_STATUS_PROTOCOL);
+ else if (parser->payload_len > self->max_payload_len)
+ ws_handler_fail (self, WS_STATUS_TOO_BIG);
+ else
+ return true;
+ return false;
+}
+
+static bool
+ws_handler_on_control_frame
+ (struct ws_handler *self, const struct ws_parser *parser)
+{
+ switch (parser->opcode)
+ {
+ case WS_OPCODE_CLOSE:
+ // TODO: confirm the close
+ break;
+ case WS_OPCODE_PING:
+ ws_handler_send_control (self, WS_OPCODE_PONG,
+ parser->input.str, parser->payload_len);
+ break;
+ case WS_OPCODE_PONG:
+ // XXX: maybe we should check the payload
+ self->received_pong = true;
+ break;
+ default:
+ // TODO: shouldn't we rather fail on unknown control frames?
+ // But should we actually return false at any time? Yes?
+ break;
+ }
+ return true;
+}
+
static bool
ws_handler_on_frame (void *user_data, const struct ws_parser *parser)
{
struct ws_handler *self = user_data;
- // TODO: handle pings and what not
- // TODO: validate the message
- // TODO: first concatenate all parts of the message
- return self->on_message (self->user_data,
+ if (ws_is_control_frame (parser->opcode))
+ return ws_handler_on_control_frame (self, parser);
+
+ // TODO: do this rather in "on_frame_header"
+ if (self->message_data.len + parser->payload_len > self->max_payload_len)
+ {
+ ws_handler_fail (self, WS_STATUS_TOO_BIG);
+ return true;
+ }
+
+ if (!self->expecting_continuation)
+ self->message_opcode = parser->opcode;
+
+ str_append_data (&self->message_data,
+ parser->input.str, parser->payload_len);
+ self->expecting_continuation = !parser->is_fin;
+
+ if (!parser->is_fin)
+ return true;
+
+ bool result = self->on_message (self->user_data, self->message_opcode,
self->parser.input.str, self->parser.payload_len);
+ str_reset (&self->message_data);
+ return result;
+}
+
+static void
+ws_handler_on_ping_timer (EV_P_ ev_timer *watcher, int revents)
+{
+ (void) loop;
+ (void) revents;
+
+ struct ws_handler *self = watcher->data;
+ if (!self->received_pong)
+ {
+ // TODO: close/fail the connection?
+ return;
+ }
+
+ ws_handler_send_control (self, WS_OPCODE_PING, NULL, 0);
}
static void
@@ -1347,7 +1498,20 @@ ws_handler_init (struct ws_handler *self)
str_init (&self->url);
ws_parser_init (&self->parser);
+ self->parser.on_frame_header = ws_handler_on_frame_header;
self->parser.on_frame = ws_handler_on_frame;
+
+ str_init (&self->message_data);
+
+ self->ping_interval = 60;
+ // This is still ridiculously high
+ self->max_payload_len = UINT32_MAX;
+
+ // Just so we can safely stop it
+ ev_timer_init (&self->ping_timer, ws_handler_on_ping_timer, 0., 0.);
+ self->ping_timer.data = self;
+ // So that the first ping timer doesn't timeout the connection
+ self->received_pong = true;
}
static void
@@ -1358,6 +1522,8 @@ ws_handler_free (struct ws_handler *self)
str_map_free (&self->headers);
str_free (&self->url);
ws_parser_free (&self->parser);
+ str_free (&self->message_data);
+ ev_timer_stop (EV_DEFAULT_ &self->ping_timer);
}
static void
@@ -1395,10 +1561,11 @@ ws_handler_on_header_value (http_parser *parser, const char *at, size_t len)
static int
ws_handler_on_headers_complete (http_parser *parser)
{
- // Just return 1 to tell the parser we don't want to parse any body;
- // the parser should have found an upgrade request for WebSockets
- (void) parser;
- return 1;
+ // We strictly require a protocol upgrade
+ if (!parser->upgrade)
+ return 2;
+
+ return 0;
}
static int
@@ -1418,6 +1585,7 @@ ws_handler_finish_handshake (struct ws_handler *self)
|| self->hp.http_major != 1
|| self->hp.http_minor != 1)
; // TODO: error (maybe send a frame depending on conditions)
+ // ...mostly just 400 Bad Request
const char *upgrade = str_map_find (&self->headers, "Upgrade");
@@ -1425,6 +1593,12 @@ ws_handler_finish_handshake (struct ws_handler *self)
const char *version = str_map_find (&self->headers, SEC_WS_VERSION);
const char *protocol = str_map_find (&self->headers, SEC_WS_PROTOCOL);
+ if (!upgrade || strcmp (upgrade, "websocket")
+ || !version || strcmp (version, "13"))
+ ; // TODO: error
+ // ... if the version doesn't match, we must send back a header indicating
+ // the version we do support
+
struct str response;
str_init (&response);
str_append (&response, "HTTP/1.1 101 Switching Protocols\r\n");
@@ -1433,8 +1607,7 @@ ws_handler_finish_handshake (struct ws_handler *self)
// TODO: prepare the rest of the headers
- // TODO: we should ideally check that this is a 16-byte base64-encoded
- // value; do we also have to strip surrounding whitespace?
+ // TODO: we should ideally check that this is a 16-byte base64-encoded value
char *response_key = ws_encode_response_key (key);
str_append_printf (&response, SEC_WS_ACCEPT ": %s\r\n", response_key);
free (response_key);
@@ -1442,6 +1615,10 @@ ws_handler_finish_handshake (struct ws_handler *self)
str_append (&response, "\r\n");
self->write_cb (self->user_data, response.str, response.len);
str_free (&response);
+
+ // XXX: maybe we should start it earlier so that the handshake can
+ // timeout as well. ws_handler_connected()?
+ ev_timer_start (EV_DEFAULT_ &self->ping_timer);
return true;
}
@@ -1477,14 +1654,16 @@ ws_handler_push (struct ws_handler *self, const void *data, size_t len)
self->state = WS_HANDLER_WEBSOCKETS;
return true;
}
- else if (n_parsed != len || HTTP_PARSER_ERRNO (&self->hp) != HPE_OK)
+
+ if (n_parsed != len || HTTP_PARSER_ERRNO (&self->hp) != HPE_OK)
{
// TODO: error
// print_debug (..., http_errno_description
// (HTTP_PARSER_ERRNO (&self->hp));
+ // NOTE: if == HPE_CB_headers_complete, "Upgrade" is missing
+ return false;
}
- // TODO: make double sure to handle the case of !upgrade
return true;
}
@@ -1497,6 +1676,7 @@ static struct config_item g_config_table[] =
{ "port_scgi", NULL, "Port to bind for SCGI" },
{ "port_ws", NULL, "Port to bind for WebSockets" },
{ "pid_file", NULL, "Full path for the PID file" },
+ // XXX: here belongs something like a web SPA that interfaces with us
{ "static_root", NULL, "The root for static content" },
{ NULL, NULL, NULL }
};
@@ -1526,11 +1706,16 @@ server_context_init (struct server_context *self)
load_config_defaults (&self->config, g_config_table);
}
+static void close_listeners (struct server_context *self);
+
static void
server_context_free (struct server_context *self)
{
- // TODO: free the clients (?)
- // TODO: close the listeners (?)
+ // We really shouldn't attempt a quit without closing the clients first
+ soft_assert (!self->clients);
+
+ close_listeners (self);
+ free (self->listeners);
str_map_free (&self->config);
}
@@ -1773,7 +1958,8 @@ struct request
{
struct server_context *ctx; ///< Server context
- void *user_data; ///< User data argument for callbacks
+ struct request_handler *handler; ///< Current request handler
+ void *handler_data; ///< User data for the handler
/// Callback to write some CGI response data to the output
void (*write_cb) (void *user_data, const void *data, size_t len);
@@ -1782,16 +1968,17 @@ struct request
/// CALLING THIS MAY CAUSE THE REQUEST TO BE DESTROYED.
void (*close_cb) (void *user_data);
- struct request_handler *handler; ///< Current request handler
- void *handler_data; ///< User data for the handler
+ void *user_data; ///< User data argument for callbacks
};
struct request_handler
{
LIST_HEADER (struct request_handler)
- /// Install ourselves as the handler for the request if applicable
- bool (*try_handle) (struct request *request, struct str_map *headers);
+ /// Install ourselves as the handler for the request if applicable.
+ /// Set @a continue_ to false if further processing should be stopped.
+ bool (*try_handle) (struct request *request,
+ struct str_map *headers, bool *continue_);
/// Handle incoming data.
/// Return false if further processing should be stopped.
@@ -1826,16 +2013,22 @@ request_finish (struct request *self)
static bool
request_start (struct request *self, struct str_map *headers)
{
+ // XXX: it feels like this should rather be two steps:
+ // bool (*can_handle) (request *, headers)
+ // ... install the handler ...
+ // bool (*handle) (request *)
+ //
+ // However that might cause some stuff to be done twice.
+ //
+ // Another way we could get rid off the continue_ argument is via adding
+ // some way of marking the request as finished from within the handler.
+
+ bool continue_ = true;
LIST_FOR_EACH (struct request_handler, handler, self->ctx->handlers)
- if (handler->try_handle (self, headers))
+ if (handler->try_handle (self, headers, &continue_))
{
- // XXX: maybe we should isolate the handlers a bit more
self->handler = handler;
-
- // TODO: we should also allow the "try_handle" function to
- // return that it has already finished processing the request
- // and we should abort it by returning false here.
- return true;
+ return continue_;
}
// Unable to serve the request
@@ -1862,7 +2055,7 @@ request_push (struct request *self, const void *data, size_t len)
static bool
request_handler_json_rpc_try_handle
- (struct request *request, struct str_map *headers)
+ (struct request *request, struct str_map *headers, bool *continue_)
{
const char *content_type = str_map_find (headers, "CONTENT_TYPE");
const char *method = str_map_find (headers, "REQUEST_METHOD");
@@ -1875,6 +2068,7 @@ request_handler_json_rpc_try_handle
str_init (buf);
request->handler_data = buf;
+ *continue_ = true;
return true;
}
@@ -1972,8 +2166,11 @@ detect_magic (const void *data, size_t len)
static bool
request_handler_static_try_handle
- (struct request *request, struct str_map *headers)
+ (struct request *request, struct str_map *headers, bool *continue_)
{
+ // Serving static files is actually quite complicated as it turns out;
+ // but this is only meant to serve a few tiny text files
+
struct server_context *ctx = request->ctx;
const char *root = str_map_find (&ctx->config, "static_root");
if (!root)
@@ -1999,6 +2196,7 @@ request_handler_static_try_handle
char *suffix = canonicalize_url_path (path_info);
char *path = xstrdup_printf ("%s%s", root, suffix);
+ // TODO: check that this is a regular file
FILE *fp = fopen (path, "rb");
if (!fp)
{
@@ -2045,6 +2243,13 @@ request_handler_static_try_handle
while ((len = fread (buf, 1, sizeof buf, fp)))
request->write_cb (request->user_data, buf, len);
fclose (fp);
+
+ // TODO: this should rather not be returned all at once but in chunks;
+ // file read requests never return EAGAIN
+ // TODO: actual file data should really be returned by a callback when
+ // the socket is writable with nothing to be sent (pumping the entire
+ // file all at once won't really work if it's huge).
+ *continue_ = false;
return true;
}
@@ -2178,7 +2383,8 @@ static void
client_fcgi_request_close (void *user_data)
{
struct client_fcgi_request *request = user_data;
- // TODO: tell the fcgi_request to what?
+ // TODO: fcgi_request_finish()? That will most probably end up with us
+ // receiving client_fcgi_request_destroy()
}
static void *
@@ -2186,6 +2392,7 @@ client_fcgi_request_start (void *user_data, struct fcgi_request *fcgi_request)
{
struct client *client = user_data;
+ // TODO: what if the request is aborted by ;
struct client_fcgi_request *request = xmalloc (sizeof *request);
request->fcgi_request = fcgi_request;
request_init (&request->request);
@@ -2375,12 +2582,17 @@ client_ws_write (void *user_data, const void *data, size_t len)
}
static bool
-client_ws_on_message (void *user_data, const void *data, size_t len)
+client_ws_on_message (void *user_data,
+ enum ws_opcode type, const void *data, size_t len)
{
struct client *client = user_data;
struct client_ws *self = client->impl_data;
- // TODO: do something about the message
+ struct str response;
+ str_init (&response);
+ process_json_rpc (client->ctx, data, len, &response);
+ // TODO: send the response
+ str_free (&response);
return true;
}
@@ -2431,6 +2643,22 @@ struct listener
struct client_impl *impl; ///< Client behaviour
};
+static void
+close_listeners (struct server_context *self)
+{
+ // TODO: factor out the closing act, to be used in initiate_quit()
+ for (size_t i = 0; i < self->n_listeners; i++)
+ {
+ struct listener *listener = &self->listeners[i];
+ if (listener->fd == -1)
+ continue;
+
+ ev_io_stop (EV_DEFAULT_ &listener->watcher);
+ xclose (listener->fd);
+ listener->fd = -1;
+ }
+}
+
static bool
client_read_loop (EV_P_ struct client *client, ev_io *watcher)
{
@@ -2472,6 +2700,8 @@ on_client_ready (EV_P_ ev_io *watcher, int revents)
// finished flushing the write queue? This should probably even be
// the default behaviour, as it's fairly uncommon for clients to
// shutdown the socket for writes while leaving it open for reading.
+ // TODO: some sort of "on_buffers_flushed" callback for streaming huge
+ // chunks of external (or generated) data.
if (!flush_queue (&client->write_queue, watcher))
goto close;
return;