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 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;
61 pa_bool_t defer_close:1;
64 static void io_callback(pa_iochannel*io, void *userdata);
65 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata);
67 pa_ioline* pa_ioline_new(pa_iochannel *io) {
71 l = pa_xnew(pa_ioline, 1);
76 l->wbuf_length = l->wbuf_index = l->wbuf_valid_length = 0;
79 l->rbuf_length = l->rbuf_index = l->rbuf_valid_length = 0;
84 l->mainloop = pa_iochannel_get_mainloop_api(io);
86 l->defer_event = l->mainloop->defer_new(l->mainloop, defer_callback, l);
87 l->mainloop->defer_enable(l->defer_event, 0);
90 l->defer_close = FALSE;
92 pa_iochannel_set_callback(io, io_callback, l);
97 static void ioline_free(pa_ioline *l) {
101 pa_iochannel_free(l->io);
104 l->mainloop->defer_free(l->defer_event);
111 void pa_ioline_unref(pa_ioline *l) {
113 pa_assert(PA_REFCNT_VALUE(l) >= 1);
115 if (PA_REFCNT_DEC(l) <= 0)
119 pa_ioline* pa_ioline_ref(pa_ioline *l) {
121 pa_assert(PA_REFCNT_VALUE(l) >= 1);
127 void pa_ioline_close(pa_ioline *l) {
129 pa_assert(PA_REFCNT_VALUE(l) >= 1);
134 pa_iochannel_free(l->io);
138 if (l->defer_event) {
139 l->mainloop->defer_free(l->defer_event);
140 l->defer_event = NULL;
147 void pa_ioline_puts(pa_ioline *l, const char *c) {
151 pa_assert(PA_REFCNT_VALUE(l) >= 1);
158 if (len > BUFFER_LIMIT - l->wbuf_valid_length)
159 len = BUFFER_LIMIT - l->wbuf_valid_length;
162 pa_assert(l->wbuf_length >= l->wbuf_valid_length);
164 /* In case the allocated buffer is too small, enlarge it. */
165 if (l->wbuf_valid_length + len > l->wbuf_length) {
166 size_t n = l->wbuf_valid_length+len;
167 char *new = pa_xnew(char, n);
170 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
177 } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
179 /* In case the allocated buffer fits, but the current index is too far from the start, move it to the front. */
180 memmove(l->wbuf, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
184 pa_assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
186 /* Append the new string */
187 memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
188 l->wbuf_valid_length += len;
190 l->mainloop->defer_enable(l->defer_event, 1);
194 void pa_ioline_set_callback(pa_ioline*l, pa_ioline_cb_t callback, void *userdata) {
196 pa_assert(PA_REFCNT_VALUE(l) >= 1);
201 l->callback = callback;
202 l->userdata = userdata;
205 static void failure(pa_ioline *l, pa_bool_t process_leftover) {
207 pa_assert(PA_REFCNT_VALUE(l) >= 1);
210 if (process_leftover && l->rbuf_valid_length > 0) {
211 /* Pass the last missing bit to the client */
214 char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
215 l->callback(l, p, l->userdata);
221 l->callback(l, NULL, l->userdata);
228 static void scan_for_lines(pa_ioline *l, size_t skip) {
230 pa_assert(PA_REFCNT_VALUE(l) >= 1);
231 pa_assert(skip < l->rbuf_valid_length);
233 while (!l->dead && l->rbuf_valid_length > skip) {
237 if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
242 p = l->rbuf + l->rbuf_index;
245 l->rbuf_index += m+1;
246 l->rbuf_valid_length -= m+1;
248 /* A shortcut for the next time */
249 if (l->rbuf_valid_length == 0)
253 l->callback(l, pa_strip_nl(p), l->userdata);
258 /* If the buffer became too large and still no newline was found, drop it. */
259 if (l->rbuf_valid_length >= BUFFER_LIMIT)
260 l->rbuf_index = l->rbuf_valid_length = 0;
263 static int do_write(pa_ioline *l);
265 static int do_read(pa_ioline *l) {
267 pa_assert(PA_REFCNT_VALUE(l) >= 1);
269 while (!l->dead && pa_iochannel_is_readable(l->io)) {
273 len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
275 /* Check if we have to enlarge the read buffer */
276 if (len < READ_SIZE) {
277 size_t n = l->rbuf_valid_length+READ_SIZE;
279 if (n >= BUFFER_LIMIT)
282 if (l->rbuf_length >= n) {
283 /* The current buffer is large enough, let's just move the data to the front */
284 if (l->rbuf_valid_length)
285 memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
287 /* Enlarge the buffer */
288 char *new = pa_xnew(char, n);
289 if (l->rbuf_valid_length)
290 memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
299 len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
301 pa_assert(len >= READ_SIZE);
304 if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
306 if (r < 0 && errno == EAGAIN)
309 if (r < 0 && errno != ECONNRESET) {
310 pa_log("read(): %s", pa_cstrerror(errno));
318 l->rbuf_valid_length += r;
320 /* Look if a line has been terminated in the newly read data */
321 scan_for_lines(l, l->rbuf_valid_length - r);
327 /* Try to flush the buffer */
328 static int do_write(pa_ioline *l) {
332 pa_assert(PA_REFCNT_VALUE(l) >= 1);
334 while (!l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length) {
336 if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
338 if (r < 0 && errno == EAGAIN)
341 if (r < 0 && errno != EPIPE)
342 pa_log("write(): %s", pa_cstrerror(errno));
350 l->wbuf_valid_length -= r;
352 /* A shortcut for the next time */
353 if (l->wbuf_valid_length == 0)
360 /* Try to flush read/write data */
361 static void do_work(pa_ioline *l) {
363 pa_assert(PA_REFCNT_VALUE(l) >= 1);
367 l->mainloop->defer_enable(l->defer_event, 0);
375 if (l->defer_close && !l->wbuf_valid_length)
381 static void io_callback(pa_iochannel*io, void *userdata) {
382 pa_ioline *l = userdata;
386 pa_assert(PA_REFCNT_VALUE(l) >= 1);
391 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
392 pa_ioline *l = userdata;
395 pa_assert(PA_REFCNT_VALUE(l) >= 1);
396 pa_assert(l->mainloop == m);
397 pa_assert(l->defer_event == e);
402 void pa_ioline_defer_close(pa_ioline *l) {
404 pa_assert(PA_REFCNT_VALUE(l) >= 1);
406 l->defer_close = TRUE;
408 if (!l->wbuf_valid_length)
409 l->mainloop->defer_enable(l->defer_event, 1);
412 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
417 pa_assert(PA_REFCNT_VALUE(l) >= 1);
419 va_start(ap, format);
420 t = pa_vsprintf_malloc(format, ap);
423 pa_ioline_puts(l, t);