text: #include <ccan/io/io.h>
    /* To reach into io_plan: not a public header! */
    #include <ccan/io/backend.h>
  #include <ccan/json_escape/json_escape.h>
  #include <ccan/json_out/json_out.h>
  #include <ccan/str/hex/hex.h>
  #include <ccan/tal/str/str.h>
  #include <common/daemon.h>
  #include <common/utils.h>
  #include <lightningd/json.h>
  #include <lightningd/json_stream.h>
  #include <lightningd/log.h>
  #include <stdarg.h>
  #include <stdio.h>
  
  struct json_stream {
  	/* NULL if we ran OOM! */
  	struct json_out *jout;
  
  	/* Who is writing to this buffer now; NULL if nobody is. */
  	struct command *writer;
  
  	/* Who is io_writing from this buffer now: NULL if nobody is. */
  	struct io_conn *reader;
  	struct io_plan *(*reader_cb)(struct io_conn *conn,
  				     struct json_stream *js,
  				     void *arg);
  	void *reader_arg;
  	size_t len_read;
  
  	/* Where to log I/O */
  	struct log *log;
  };
  
  static void adjust_io_write(struct json_out *jout,
  			    ptrdiff_t delta,
  			    struct json_stream *js)
  {
  	/* If io_write is in progress, we shift it to point to new buffer pos */
  	if (js->reader)
  	/* FIXME: This, or something prettier (io_replan?) belong in ccan/io! */
  		js->reader->plan[IO_OUT].arg.u1.cp += delta;
  }
  
  struct json_stream *new_json_stream(const tal_t *ctx,
  				    struct command *writer,
  				    struct log *log)
  {
  	struct json_stream *js = tal(ctx, struct json_stream);
  
  	/* FIXME: Add magic so tal_resize can fail! */
  	js->jout = json_out_new(js);
  	json_out_call_on_move(js->jout, adjust_io_write, js);
  	js->writer = writer;
  	js->reader = NULL;
  	js->log = log;
  	return js;
  }
  
  struct json_stream *json_stream_dup(const tal_t *ctx,
  				    struct json_stream *original,
  				    struct log *log)
  {
  	struct json_stream *js = tal_dup(ctx, struct json_stream, original);
  
  	if (original->jout)
  		js->jout = json_out_dup(js, original->jout);
  	js->log = log;
  	return js;
  }
  
  bool json_stream_still_writing(const struct json_stream *js)
  {
  	return js->writer != NULL;
  }
  
  void json_stream_log_suppress(struct json_stream *js, const char *cmd_name)
  {
  	/* Really shouldn't be used for anything else */
  	assert(streq(cmd_name, "getlog"));
  	js->log = NULL;
  }
  
  /* If we have an allocation failure. */
  static void COLD js_oom(struct json_stream *js)
  {
  	js->jout = tal_free(js->jout);
  }
  
  void json_stream_append(struct json_stream *js,
  			const char *str, size_t len)
  {
  	char *dest;
  
  	if (!js->jout)
  		return;
  	dest = json_out_direct(js->jout, len);
  	if (!dest) {
  		js_oom(js);
  		return;
  	}
  	memcpy(dest, str, len);
  }
  
  void json_stream_close(struct json_stream *js, struct command *writer)
  {
  	/* FIXME: We use writer == NULL for malformed: make writer a void *?
  	 * I used to assert(writer); here. */
  	assert(js->writer == writer);
  
  	/* Should be well-formed at this point! */
  	json_out_finished(js->jout);
  	json_stream_append(js, "\n\n", strlen("\n\n"));
  	json_stream_flush(js);
  	js->writer = NULL;
  }
  
  /* Also called when we're oom, so it will kill reader. */
  void json_stream_flush(struct json_stream *js)
  {
  	/* Wake the stream reader. FIXME:  Could have a flag here to optimize */
  	io_wake(js);
  }
  
  char *json_member_direct(struct json_stream *js,
  			 const char *fieldname, size_t extra)
  {
  	char *dest;
  
  	if (!js->jout)
  		return NULL;
  
  	dest = json_out_member_direct(js->jout, fieldname, extra);
  	if (!dest)
  		js_oom(js);
  	return dest;
  }
  
  void json_array_start(struct json_stream *js, const char *fieldname)
  {
  	if (js->jout && !json_out_start(js->jout, fieldname, '['))
  		js_oom(js);
  }
  
  void json_array_end(struct json_stream *js)
  {
  	if (js->jout && !json_out_end(js->jout, ']'))
  		js_oom(js);
  }
  
  void json_object_start(struct json_stream *js, const char *fieldname)
  {
  	if (js->jout && !json_out_start(js->jout, fieldname, '{'))
  		js_oom(js);
  }
  
  void json_object_end(struct json_stream *js)
  {
  	if (js->jout && !json_out_end(js->jout, '}'))
  		js_oom(js);
  }
  
  void json_object_compat_end(struct json_stream *js)
  {
  	/* In 0.7.1 we upgraded pylightning to no longer need this. */
  #ifdef COMPAT_V070
  	json_stream_append(js, " ", 1);
  #endif
  	json_object_end(js);
  }
  
  void json_add_member(struct json_stream *js,
  		     const char *fieldname,
  		     bool quote,
  		     const char *fmt, ...)
  {
  	va_list ap;
  
  	va_start(ap, fmt);
  	if (js->jout && !json_out_addv(js->jout, fieldname, quote, fmt, ap))
  		js_oom(js);
  	va_end(ap);
  }
  
  /* This is where we read the json_stream and write it to conn */
  static struct io_plan *json_stream_output_write(struct io_conn *conn,
  						struct json_stream *js)
  {
  	const char *p;
  
  	/* Out of memory?  Nothing we can do but close conn */
  	if (!js->jout)
  		return io_close(conn);
  
  	/* For when we've just done some output */
  	json_out_consume(js->jout, js->len_read);
  
  	/* Get how much we can write out from js */
  	p = json_out_contents(js->jout, &js->len_read);
  
  	/* Nothing in buffer? */
  	if (!p) {
  		/* We're not doing io_write now, unset. */
  		js->reader = NULL;
  		if (!json_stream_still_writing(js))
  			return js->reader_cb(conn, js, js->reader_arg);
  		return io_out_wait(conn, js, json_stream_output_write, js);
  	}
  
  	js->reader = conn;
  	if (js->log)
  		log_io(js->log, LOG_IO_OUT, "", p, js->len_read);
  	return io_write(conn,
  			p, js->len_read,
  			json_stream_output_write, js);
  }
  
  struct io_plan *json_stream_output_(struct json_stream *js,
  				    struct io_conn *conn,
  				    struct io_plan *(*cb)(struct io_conn *conn,
  							  struct json_stream *js,
  							  void *arg),
  				    void *arg)
  {
  	assert(!js->reader);
  
  	js->reader_cb = cb;
  	js->reader_arg = arg;
  
  	js->len_read = 0;
  	return json_stream_output_write(conn, js);
  }
  