static void enable_mainloop_sources(struct pa_iochannel *io) {
assert(io);
- if (io->input_event == io->output_event) {
+ if (io->input_event == io->output_event && io->input_event) {
enum pa_io_event_flags f = PA_IO_EVENT_NULL;
assert(io->input_event);
if ((f & (PA_IO_EVENT_HANGUP|PA_IO_EVENT_ERROR)) && !io->hungup) {
io->hungup = 1;
changed = 1;
- }
-
- if ((f & PA_IO_EVENT_INPUT) && !io->readable) {
- io->readable = 1;
- changed = 1;
- assert(e == io->input_event);
- }
-
- if ((f & PA_IO_EVENT_OUTPUT) && !io->writable) {
- io->writable = 1;
- changed = 1;
- assert(e == io->output_event);
+
+ if (e == io->input_event) {
+ io->mainloop->io_free(io->input_event);
+ io->input_event = NULL;
+ }
+
+ if (e == io->output_event) {
+ io->mainloop->io_free(io->output_event);
+ io->output_event = NULL;
+ }
+ } else {
+
+ if ((f & PA_IO_EVENT_INPUT) && !io->readable) {
+ io->readable = 1;
+ changed = 1;
+ assert(e == io->input_event);
+ }
+
+ if ((f & PA_IO_EVENT_OUTPUT) && !io->writable) {
+ io->writable = 1;
+ changed = 1;
+ assert(e == io->output_event);
+ }
}
if (changed) {
int pa_iochannel_is_readable(struct pa_iochannel*io) {
assert(io);
- return io->readable;
+ return io->readable || io->hungup;
}
int pa_iochannel_is_writable(struct pa_iochannel*io) {
assert(io);
- return io->writable;
+ return io->writable && !io->hungup;
}
int pa_iochannel_is_hungup(struct pa_iochannel*io) {
ssize_t pa_iochannel_write(struct pa_iochannel*io, const void*data, size_t l) {
ssize_t r;
- assert(io);
- assert(data);
- assert(l);
- assert(io->ofd >= 0);
-
-
assert(io && data && l && io->ofd >= 0);
-
-
if ((r = write(io->ofd, data, l)) >= 0) {
io->writable = 0;
enable_mainloop_sources(io);
ssize_t pa_iochannel_read(struct pa_iochannel*io, void*data, size_t l) {
ssize_t r;
-
assert(io && data && io->ifd >= 0);
if ((r = read(io->ifd, data, l)) >= 0) {
return pa_socket_set_sndbuf(io->ofd, l);
}
-void pa_iochannel_force_unreadable(struct pa_iochannel *io) {
- assert(io);
- io->readable = 0;
- enable_mainloop_sources(io);
-}
-
-void pa_iochannel_force_unwritable(struct pa_iochannel *io) {
- io->writable = 0;
- enable_mainloop_sources(io);
-}
int pa_iochannel_is_writable(struct pa_iochannel*io);
int pa_iochannel_is_hungup(struct pa_iochannel*io);
-void pa_iochannel_force_unreadable(struct pa_iochannel *io);
-void pa_iochannel_force_unwritable(struct pa_iochannel *io);
-
void pa_iochannel_set_noclose(struct pa_iochannel*io, int b);
void pa_iochannel_set_callback(struct pa_iochannel*io, void (*callback)(struct pa_iochannel*io, void *userdata), void *userdata);
pa_xfree(c->read_data);
pa_xfree(c->write_data);
-
- pa_iochannel_free(c->io);
+
+ if (c->io)
+ pa_iochannel_free(c->io);
if (c->defer_event)
c->protocol->core->mainloop->defer_free(c->defer_event);
static int do_read(struct connection *c) {
assert(c && c->io);
+/* pa_log("READ\n"); */
+
if (c->state == ESD_NEXT_REQUEST) {
ssize_t r;
assert(c->read_data_length < sizeof(c->request));
handler = proto_map+c->request;
- pa_log(__FILE__": executing request #%u\n", c->request);
+/* pa_log(__FILE__": executing request #%u\n", c->request); */
if (!handler->proc) {
pa_log(__FILE__": recieved unimplemented request #%u.\n", c->request);
assert(c->input_memblockq);
+/* pa_log("STREAMING_DATA\n"); */
+
if (!(l = pa_memblockq_missing(c->input_memblockq)))
return 0;
pa_log(__FILE__": read() failed: %s\n", r == 0 ? "EOF" : strerror(errno));
return -1;
}
+
+/* pa_log(__FILE__": read %u\n", r); */
chunk.memblock = c->playback.current_memblock;
chunk.index = c->playback.memblock_index;
static int do_write(struct connection *c) {
assert(c && c->io);
+/* pa_log("WRITE\n"); */
+
if (c->write_data_length) {
ssize_t r;
assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable);
c->protocol->core->mainloop->defer_enable(c->defer_event, 0);
+
+/* pa_log("DOWORK\n"); */
- if (c->dead)
+ if (c->dead || !c->io)
return;
-
+
if (pa_iochannel_is_readable(c->io))
if (do_read(c) < 0)
goto fail;
-
+
if (pa_iochannel_is_writable(c->io))
if (do_write(c) < 0)
goto fail;
-
+
return;
fail:
if (c->state == ESD_STREAMING_DATA && c->sink_input) {
c->dead = 1;
pa_memblockq_prebuf_disable(c->input_memblockq);
- c->protocol->core->mainloop->defer_enable(c->defer_event, 0);
+
+ pa_iochannel_free(c->io);
+ c->io = NULL;
+
} else
connection_free(c);
}
struct connection *c = userdata;
assert(io && c && c->io == io);
+/* pa_log("IO\n"); */
+
do_work(c);
}
struct connection *c = userdata;
assert(a && c && c->defer_event == e);
+/* pa_log("DEFER\n"); */
+
do_work(c);
}
struct connection*c = i->userdata;
assert(i && c && length);
+/* pa_log("DROP\n"); */
+
pa_memblockq_drop(c->input_memblockq, chunk, length);
/* do something */