evirc: comparison src/line

equal deleted inserted replaced

-:dedf137b504f
+:8c80580ccde9
 */
 struct line_proto {
 /* The sock_stream we read/write with */
 struct sock_stream *sock;
-/* The incoming line buffer */
+/* The incoming/outgoing line buffer */
-char *buf;
+char *in_buf, *out_buf;
-/* Incoming buffer size */
+/* Buffer size (same for both) */
 size_t buf_len;
 /* Offset of trailing data in buf */
 size_t tail_offset;
 /* Length of trailing data in buf, if any */
 size_t tail_len;
+/* Amount of data in the out buffer */
+size_t out_offset;
 /* Last error */
 struct error_info err;
 /* Callback info */
 // sanity-check
 assert(lp->tail_offset < lp->buf_len);
 do {
 // attempt to read a line
-if (line_proto_read(lp, &line))
+if (line_proto_recv(lp, &line))
 // XXX: fail
-errx(1, "line_proto_read: %s", error_msg(&lp->err));
+errx(1, "line_proto_recv: %s", error_msg(&lp->err));
 // got a line?
 if (line)
 lp->cb_read(line, lp->cb_arg);
 if (line_proto_schedule_events(lp, EV_READ))
 errx(1, "line_proto_schedule_events: %s", error_msg(&lp->err));
 }
 /*
+* Signal for write
+*/
+static void line_proto_on_write (struct sock_stream *sock, void *arg)
+{
+struct line_proto *lp = arg;
+err_t err;
+// just flush
+if ((err = line_proto_flush(lp)) < 0)
+errx(1, "line_proto_flush: %s", error_name(err));
+}
+/*
 * Schedule our sock_stream callback
 */
 static err_t line_proto_schedule_events (struct line_proto *lp, short what)
 {
 // just use sock_stream's interface
 err_t line_proto_create (struct line_proto **lp_ptr, struct sock_stream *sock, size_t buf_size,
 line_proto_read_cb cb_func, void *cb_arg, struct error_info *err)
 {
 struct line_proto *lp;
-// allocate
+// allocate struct and buffers
-if ((lp = calloc(1, sizeof(*lp))) == NULL)
+if (
-return SET_ERROR(err, ERR_CALLOC);
+(lp = calloc(1, sizeof(*lp))) == NULL
+||  (lp->in_buf = malloc(buf_size)) == NULL
-// allocate buffer
+||  (lp->out_buf = malloc(buf_size)) == NULL
-if ((lp->buf = malloc(buf_size)) == NULL) {
+)
-free(lp);
+JUMP_SET_ERROR(err, ERR_CALLOC);
-return SET_ERROR(err, ERR_CALLOC);
-}
 // store
 lp->sock = sock;
 lp->buf_len = buf_size;
 lp->cb_read = cb_func;
 // return
 *lp_ptr = lp;
 return SUCCESS;
+error:
+if (lp) {
+free(lp->in_buf);
+free(lp->out_buf);
+}
+free(lp);
 }
 /*
 * This looks for a full '\r\n' terminated line at the beginning of the given buffer. If found, the \r\n will be
 * replaced with a '\0', and the offset to the beginning of the next line returned. If not found, zero is returned
 // return offset to next line, as set in loop based on delim
 return next;
 }
-err_t line_proto_read (struct line_proto *lp, char **line_ptr)
+err_t line_proto_recv (struct line_proto *lp, char **line_ptr)
 {
 // offset to recv() new data into, offset to _parse_line hint, offset to next line from _parse_line
 size_t recv_offset = 0, peek_offset = 0, next_offset = 0;
 int ret;
 recv_offset = lp->tail_len;
 // move trailing data from previous line to front of buffer
 if (lp->tail_offset) {
 // move to front
-memmove(lp->buf, lp->buf + lp->tail_offset, lp->tail_len);
+memmove(lp->in_buf, lp->in_buf + lp->tail_offset, lp->tail_len);
 // reset
 lp->tail_offset = 0;
 lp->tail_len = 0;
 }
 // readline loop
 do {
 // parse any line at the beginning of the buffer
-if ((next_offset = _parse_line(lp->buf, recv_offset, &peek_offset)) > 0) {
+if ((next_offset = _parse_line(lp->in_buf, recv_offset, &peek_offset)) > 0) {
 // store a valid *line_ptr
-*line_ptr = lp->buf;
+*line_ptr = lp->in_buf;
 // exit loop and return
 break;
 }
 // ensure there's enough space for the rest of the line
-// XXX: this must be an error, not an assert
+if (recv_offset >= lp->buf_len)
-assert(recv_offset < lp->buf_len);
+return ERR_LINE_TOO_LONG;
 // otherwise, read more data
-if ((ret = sock_stream_read(lp->sock, lp->buf + recv_offset, lp->buf_len - recv_offset)) < 0)
+if ((ret = sock_stream_read(lp->sock, lp->in_buf + recv_offset, lp->buf_len - recv_offset)) < 0)
 // store and return NULL on errors
 RETURN_SET_ERROR_INFO(&lp->err, sock_stream_error(lp->sock));
 // EAGAIN?
 if (ret == 0) {
 // ok
 return SUCCESS;
 }
-int line_proto_write (struct line_proto *lp, const char *line)
+int line_proto_send (struct line_proto *lp, const char *line)
 {
 int ret;
 size_t len = strlen(line);
-// XXX: no output buffers for now :)
+// drop line if we already have output buffered
-if ((ret = sock_stream_write(lp->sock, line, len)) < 0)
+if (lp->out_offset)
-RETURN_SET_ERROR_INFO(&lp->err, sock_stream_error(lp->sock));
+return -ERR_WRITE_EOF;
+// try and write the line
+if ((ret = sock_stream_write(lp->sock, line, len)) < 0) {
+SET_ERROR_INFO(&lp->err, sock_stream_error(lp->sock));
+return -ERROR_CODE(&lp->err);
+}
 // EAGAIN or partial?
 if (ret < len) {
-// XXX: ugly hack, need partial line buffering
+size_t trailing = len - ret;
-return -1;
-}
+// ensure it's not waaaay too long
+if (trailing > lp->buf_len)
+return -ERR_LINE_TOO_LONG;
+// copy remaining portion to buffer
+memcpy(lp->out_buf, line + ret, trailing);
+// update offset
+lp->out_offset = trailing;
+// register for EV_WRITE
+line_proto_schedule_events(lp, EV_READ | EV_WRITE);
+// buffered...
+return 1;
+} else {
+// ok, no buffering needed
+return SUCCESS;
+}
+}
+int line_proto_flush (struct line_proto *lp)
+{
+int ret;
+// try and write the line
+if ((ret = sock_stream_write(lp->sock, lp->out_buf, lp->out_offset)) < 0) {
+SET_ERROR_INFO(&lp->err, sock_stream_error(lp->sock));
+return -ERROR_CODE(&lp->err);
+}
+// empty now?
+if (ret == lp->out_offset) {
+lp->out_offset = 0;
+return SUCCESS;
+}
+// partial?
+if (ret > 0) {
+size_t remaining = lp->out_offset - ret;
+// move the rest up
+memmove(lp->out_buf, lp->out_buf + ret, remaining);
+// update offset
+lp->out_offset = remaining;
+}
+// register for EV_WRITE
+line_proto_schedule_events(lp, EV_READ | EV_WRITE);
 // ok
-return SUCCESS;
+return 1;
 }
 const struct error_info* line_proto_error (struct line_proto *lp)
 {
 // return pointer
 return &lp->err;
 }

changeset 19	8c80580ccde9
parent 18	dedf137b504f
child 20	d9c4c2980a0d