#include <stdio.h>
#include "ustream.h"
-static void ustream_fd_set_uloop(struct ustream *s)
+static void ustream_fd_set_uloop(struct ustream *s, bool write)
{
struct ustream_fd *sf = container_of(s, struct ustream_fd, stream);
struct ustream_buf *buf;
flags |= ULOOP_READ;
buf = s->w.head;
- if (buf && s->w.data_bytes && !s->write_error)
+ if (write || (buf && s->w.data_bytes && !s->write_error))
flags |= ULOOP_WRITE;
uloop_fd_add(&sf->fd, flags);
sf->fd.cb(&sf->fd, ULOOP_READ);
}
+static void ustream_fd_set_read_blocked(struct ustream *s)
+{
+ ustream_fd_set_uloop(s, false);
+}
+
static void ustream_fd_read_pending(struct ustream_fd *sf, bool *more)
{
struct ustream *s = &sf->stream;
goto retry;
if (errno == EAGAIN || errno == EWOULDBLOCK)
- return 0;
+ len = 0;
}
+ if (len >= 0 && len < buflen)
+ ustream_fd_set_uloop(s, true);
+
return len;
}
if (events & ULOOP_WRITE) {
if (!ustream_write_pending(s))
- ustream_fd_set_uloop(s);
+ ustream_fd_set_uloop(s, false);
}
if (!s->eof && fd->eof) {
s->eof = true;
- ustream_fd_set_uloop(s);
+ ustream_fd_set_uloop(s, false);
ustream_state_change(s);
}
sf->fd.fd = fd;
sf->fd.cb = ustream_uloop_cb;
- s->set_read_blocked = ustream_fd_set_uloop;
+ s->set_read_blocked = ustream_fd_set_read_blocked;
s->write = ustream_fd_write;
s->free = ustream_fd_free;
s->poll = ustream_fd_poll;
- ustream_fd_set_uloop(s);
+ ustream_fd_set_uloop(s, false);
}