json: Support streaming JSON messages

It turns out we were heavily relying on the fact that after each message from
the client there'd be a flush, and that there would not be anything after the
JSON object we read. This will no longer be the case once we start streaming
things or we are very quick in issuing the JSON-RPC requests.

This just takes one of the error paths (incomplete read) and makes it into a
successful path if we have indeed read a full root element.

common/json.c

@@ -187,7 +187,10 @@ jsmntok_t *json_parse_input(const char *input, int len, bool *valid)
 	jsmntok_t *toks;
 	int ret;
 
-	toks = tal_arr(input, jsmntok_t, 10);
+	/* Zero out so we can count elements correctly even on incomplete reads
+	 * (when jsmn_parse returns -1). This results in all toks being of type
+	 * JSMN_UNDEFINED which we can recognize. */
+	toks = tal_arrz(input, jsmntok_t, 10);
 
 again:
 	jsmn_init(&parser);
@@ -197,14 +200,27 @@ again:
 	case JSMN_ERROR_INVAL:
 		*valid = false;
 		return tal_free(toks);
-	case JSMN_ERROR_PART:
-		*valid = true;
-		return tal_free(toks);
 	case JSMN_ERROR_NOMEM:
-		tal_resize(&toks, tal_count(toks) * 2);
+		tal_resizez(&toks, tal_count(toks) * 2);
 		goto again;
 	}
 
+	/* Check whether we read at least one full root element, i.e., root
+	 * element has its end set. */
+	if (toks[0].type == JSMN_UNDEFINED || toks[0].end == -1) {
+		*valid = true;
+		return tal_free(toks);
+	}
+
+	/* If we read a partial element at the end of the stream we'll get a
+	 * ret=JSMN_ERROR_PART, but due to the previous check we know we read at
+	 * least one full element, so count tokens that are part of this root
+	 * element. */
+	for (ret=0; ret < tal_count(toks)-1; ret++) {
+		if (toks[ret].type == JSMN_UNDEFINED || toks[ret].start >= toks[0].end)
+			break;
+	}
+
 	/* Cut to length and return. */
 	*valid = true;
 	tal_resize(&toks, ret + 1);

common/test/run-json.c

@@ -145,6 +145,37 @@ static void test_json_partial(void)
 	tal_free(ctx);
 }
 
+/* Test that we can segment and parse a stream of json objects correctly. */
+static void test_json_stream(void)
+{
+	bool valid;
+	char *input, *talstr;
+	jsmntok_t *toks;
+
+	/* Multiple full messages in a single buffer (happens when buffer
+	 * boundary coincides with message boundary, or read returned after
+	 * timeout. */
+	input = "{\"x\":\"x\"}{\"y\":\"y\"}";
+	talstr = tal_strndup(NULL, input, strlen(input));
+	toks = json_parse_input(talstr, strlen(talstr), &valid);
+	assert(toks);
+	assert(tal_count(toks) == 4);
+	assert(toks[0].start == 0 && toks[0].end == 9);
+	assert(valid);
+	tal_free(talstr);
+
+	/* Multiple messages, and the last one is partial, far more likely than
+	 * accidentally getting the boundaries to match. */
+	input = "{\"x\":\"x\"}{\"y\":\"y\"}{\"z\":\"z";
+	talstr = tal_strndup(NULL, input, strlen(input));
+	toks = json_parse_input(talstr, strlen(talstr), &valid);
+	assert(toks);
+	assert(tal_count(toks) == 4);
+	assert(toks[0].start == 0 && toks[0].end == 9);
+	assert(valid);
+	tal_free(talstr);
+}
+
 int main(void)
 {
 	setup_locale();
@@ -153,6 +184,7 @@ int main(void)
 	test_json_filter();
 	test_json_escape();
 	test_json_partial();
+	test_json_stream();
 	assert(!taken_any());
 	take_cleanup();
 }