2 This file is part of PulseAudio.
4 Copyright 2004-2006 Lennart Poettering
6 PulseAudio is free software; you can redistribute it and/or modify
7 it under the terms of the GNU Lesser General Public License as published
8 by the Free Software Foundation; either version 2.1 of the License,
9 or (at your option) any later version.
11 PulseAudio is distributed in the hope that it will be useful, but
12 WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 General Public License for more details.
16 You should have received a copy of the GNU Lesser General Public License
17 along with PulseAudio; if not, write to the Free Software
18 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
31 #include <pulse/xmalloc.h>
33 #include <pulsecore/winsock.h>
34 #include <pulsecore/core-error.h>
35 #include <pulsecore/log.h>
36 #include <pulsecore/macro.h>
37 #include <pulsecore/refcnt.h>
41 #define BUFFER_LIMIT (64*1024)
42 #define READ_SIZE (1024)
48 pa_defer_event *defer_event;
49 pa_mainloop_api *mainloop;
52 size_t wbuf_length, wbuf_index, wbuf_valid_length;
55 size_t rbuf_length, rbuf_index, rbuf_valid_length;
57 pa_ioline_cb_t callback;
60 pa_ioline_drain_cb_t drain_callback;
64 pa_bool_t defer_close:1;
67 static void io_callback(pa_iochannel*io, void *userdata);
68 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata);
70 pa_ioline* pa_ioline_new(pa_iochannel *io) {
74 l = pa_xnew(pa_ioline, 1);
79 l->wbuf_length = l->wbuf_index = l->wbuf_valid_length = 0;
82 l->rbuf_length = l->rbuf_index = l->rbuf_valid_length = 0;
87 l->drain_callback = NULL;
88 l->drain_userdata = NULL;
90 l->mainloop = pa_iochannel_get_mainloop_api(io);
92 l->defer_event = l->mainloop->defer_new(l->mainloop, defer_callback, l);
93 l->mainloop->defer_enable(l->defer_event, 0);
96 l->defer_close = FALSE;
98 pa_iochannel_set_callback(io, io_callback, l);
103 static void ioline_free(pa_ioline *l) {
107 pa_iochannel_free(l->io);
110 l->mainloop->defer_free(l->defer_event);
117 void pa_ioline_unref(pa_ioline *l) {
119 pa_assert(PA_REFCNT_VALUE(l) >= 1);
121 if (PA_REFCNT_DEC(l) <= 0)
125 pa_ioline* pa_ioline_ref(pa_ioline *l) {
127 pa_assert(PA_REFCNT_VALUE(l) >= 1);
133 void pa_ioline_close(pa_ioline *l) {
135 pa_assert(PA_REFCNT_VALUE(l) >= 1);
140 pa_iochannel_free(l->io);
144 if (l->defer_event) {
145 l->mainloop->defer_free(l->defer_event);
146 l->defer_event = NULL;
153 void pa_ioline_puts(pa_ioline *l, const char *c) {
157 pa_assert(PA_REFCNT_VALUE(l) >= 1);
164 if (len > BUFFER_LIMIT - l->wbuf_valid_length)
165 len = BUFFER_LIMIT - l->wbuf_valid_length;
168 pa_assert(l->wbuf_length >= l->wbuf_valid_length);
170 /* In case the allocated buffer is too small, enlarge it. */
171 if (l->wbuf_valid_length + len > l->wbuf_length) {
172 size_t n = l->wbuf_valid_length+len;
173 char *new = pa_xnew(char, (unsigned) n);
176 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
183 } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
185 /* In case the allocated buffer fits, but the current index is too far from the start, move it to the front. */
186 memmove(l->wbuf, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
190 pa_assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
192 /* Append the new string */
193 memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
194 l->wbuf_valid_length += len;
196 l->mainloop->defer_enable(l->defer_event, 1);
200 void pa_ioline_set_callback(pa_ioline*l, pa_ioline_cb_t callback, void *userdata) {
202 pa_assert(PA_REFCNT_VALUE(l) >= 1);
207 l->callback = callback;
208 l->userdata = userdata;
211 void pa_ioline_set_drain_callback(pa_ioline*l, pa_ioline_drain_cb_t callback, void *userdata) {
213 pa_assert(PA_REFCNT_VALUE(l) >= 1);
218 l->drain_callback = callback;
219 l->drain_userdata = userdata;
222 static void failure(pa_ioline *l, pa_bool_t process_leftover) {
224 pa_assert(PA_REFCNT_VALUE(l) >= 1);
227 if (process_leftover && l->rbuf_valid_length > 0) {
228 /* Pass the last missing bit to the client */
231 char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
232 l->callback(l, p, l->userdata);
238 l->callback(l, NULL, l->userdata);
245 static void scan_for_lines(pa_ioline *l, size_t skip) {
247 pa_assert(PA_REFCNT_VALUE(l) >= 1);
248 pa_assert(skip < l->rbuf_valid_length);
250 while (!l->dead && l->rbuf_valid_length > skip) {
254 if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
259 p = l->rbuf + l->rbuf_index;
262 l->rbuf_index += m+1;
263 l->rbuf_valid_length -= m+1;
265 /* A shortcut for the next time */
266 if (l->rbuf_valid_length == 0)
270 l->callback(l, pa_strip_nl(p), l->userdata);
275 /* If the buffer became too large and still no newline was found, drop it. */
276 if (l->rbuf_valid_length >= BUFFER_LIMIT)
277 l->rbuf_index = l->rbuf_valid_length = 0;
280 static int do_write(pa_ioline *l);
282 static int do_read(pa_ioline *l) {
284 pa_assert(PA_REFCNT_VALUE(l) >= 1);
286 while (l->io && !l->dead && pa_iochannel_is_readable(l->io)) {
290 len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
292 /* Check if we have to enlarge the read buffer */
293 if (len < READ_SIZE) {
294 size_t n = l->rbuf_valid_length+READ_SIZE;
296 if (n >= BUFFER_LIMIT)
299 if (l->rbuf_length >= n) {
300 /* The current buffer is large enough, let's just move the data to the front */
301 if (l->rbuf_valid_length)
302 memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
304 /* Enlarge the buffer */
305 char *new = pa_xnew(char, (unsigned) n);
306 if (l->rbuf_valid_length)
307 memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
316 len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
318 pa_assert(len >= READ_SIZE);
321 if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
323 if (r < 0 && errno == EAGAIN)
326 if (r < 0 && errno != ECONNRESET) {
327 pa_log("read(): %s", pa_cstrerror(errno));
335 l->rbuf_valid_length += (size_t) r;
337 /* Look if a line has been terminated in the newly read data */
338 scan_for_lines(l, l->rbuf_valid_length - (size_t) r);
344 /* Try to flush the buffer */
345 static int do_write(pa_ioline *l) {
349 pa_assert(PA_REFCNT_VALUE(l) >= 1);
351 while (l->io && !l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length > 0) {
353 if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
355 if (r < 0 && errno == EAGAIN)
358 if (r < 0 && errno != EPIPE)
359 pa_log("write(): %s", pa_cstrerror(errno));
366 l->wbuf_index += (size_t) r;
367 l->wbuf_valid_length -= (size_t) r;
369 /* A shortcut for the next time */
370 if (l->wbuf_valid_length == 0)
374 if (l->wbuf_valid_length <= 0 && l->drain_callback)
375 l->drain_callback(l, l->drain_userdata);
380 /* Try to flush read/write data */
381 static void do_work(pa_ioline *l) {
383 pa_assert(PA_REFCNT_VALUE(l) >= 1);
387 l->mainloop->defer_enable(l->defer_event, 0);
395 if (l->defer_close && !l->wbuf_valid_length)
401 static void io_callback(pa_iochannel*io, void *userdata) {
402 pa_ioline *l = userdata;
406 pa_assert(PA_REFCNT_VALUE(l) >= 1);
411 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
412 pa_ioline *l = userdata;
415 pa_assert(PA_REFCNT_VALUE(l) >= 1);
416 pa_assert(l->mainloop == m);
417 pa_assert(l->defer_event == e);
422 void pa_ioline_defer_close(pa_ioline *l) {
424 pa_assert(PA_REFCNT_VALUE(l) >= 1);
426 l->defer_close = TRUE;
428 if (!l->wbuf_valid_length)
429 l->mainloop->defer_enable(l->defer_event, 1);
432 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
437 pa_assert(PA_REFCNT_VALUE(l) >= 1);
439 va_start(ap, format);
440 t = pa_vsprintf_malloc(format, ap);
443 pa_ioline_puts(l, t);
447 pa_iochannel* pa_ioline_detach_iochannel(pa_ioline *l) {
458 pa_iochannel_set_callback(r, NULL, NULL);
463 pa_bool_t pa_ioline_is_drained(pa_ioline *l) {
466 return l->wbuf_valid_length <= 0;