don't fail if a signalled writability of STDOUT is no longer true when we try it...
[platform/upstream/pulseaudio.git] / src / pulsecore / ioline.c
1 /* $Id$ */
2
3 /***
4   This file is part of PulseAudio.
5
6   Copyright 2004-2006 Lennart Poettering
7
8   PulseAudio is free software; you can redistribute it and/or modify
9   it under the terms of the GNU Lesser General Public License as published
10   by the Free Software Foundation; either version 2 of the License,
11   or (at your option) any later version.
12
13   PulseAudio is distributed in the hope that it will be useful, but
14   WITHOUT ANY WARRANTY; without even the implied warranty of
15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16   General Public License for more details.
17
18   You should have received a copy of the GNU Lesser General Public License
19   along with PulseAudio; if not, write to the Free Software
20   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
21   USA.
22 ***/
23
24 #ifdef HAVE_CONFIG_H
25 #include <config.h>
26 #endif
27
28 #include <errno.h>
29 #include <stdio.h>
30 #include <stdlib.h>
31 #include <string.h>
32
33 #include <pulse/xmalloc.h>
34
35 #include <pulsecore/winsock.h>
36 #include <pulsecore/core-error.h>
37 #include <pulsecore/log.h>
38 #include <pulsecore/macro.h>
39 #include <pulsecore/refcnt.h>
40
41 #include "ioline.h"
42
43 #define BUFFER_LIMIT (64*1024)
44 #define READ_SIZE (1024)
45
46 struct pa_ioline {
47     PA_REFCNT_DECLARE;
48
49     pa_iochannel *io;
50     pa_defer_event *defer_event;
51     pa_mainloop_api *mainloop;
52     int dead;
53
54     char *wbuf;
55     size_t wbuf_length, wbuf_index, wbuf_valid_length;
56
57     char *rbuf;
58     size_t rbuf_length, rbuf_index, rbuf_valid_length;
59
60     void (*callback)(pa_ioline*io, const char *s, void *userdata);
61     void *userdata;
62
63     int defer_close;
64 };
65
66 static void io_callback(pa_iochannel*io, void *userdata);
67 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata);
68
69 pa_ioline* pa_ioline_new(pa_iochannel *io) {
70     pa_ioline *l;
71     pa_assert(io);
72
73     l = pa_xnew(pa_ioline, 1);
74     PA_REFCNT_INIT(l);
75     l->io = io;
76     l->dead = 0;
77
78     l->wbuf = NULL;
79     l->wbuf_length = l->wbuf_index = l->wbuf_valid_length = 0;
80
81     l->rbuf = NULL;
82     l->rbuf_length = l->rbuf_index = l->rbuf_valid_length = 0;
83
84     l->callback = NULL;
85     l->userdata = NULL;
86
87     l->mainloop = pa_iochannel_get_mainloop_api(io);
88
89     l->defer_event = l->mainloop->defer_new(l->mainloop, defer_callback, l);
90     l->mainloop->defer_enable(l->defer_event, 0);
91
92     l->defer_close = 0;
93
94     pa_iochannel_set_callback(io, io_callback, l);
95
96     return l;
97 }
98
99 static void ioline_free(pa_ioline *l) {
100     pa_assert(l);
101
102     if (l->io)
103         pa_iochannel_free(l->io);
104
105     if (l->defer_event)
106         l->mainloop->defer_free(l->defer_event);
107
108     pa_xfree(l->wbuf);
109     pa_xfree(l->rbuf);
110     pa_xfree(l);
111 }
112
113 void pa_ioline_unref(pa_ioline *l) {
114     pa_assert(l);
115     pa_assert(PA_REFCNT_VALUE(l) >= 1);
116
117     if (PA_REFCNT_DEC(l) <= 0)
118         ioline_free(l);
119 }
120
121 pa_ioline* pa_ioline_ref(pa_ioline *l) {
122     pa_assert(l);
123     pa_assert(PA_REFCNT_VALUE(l) >= 1);
124
125     PA_REFCNT_INC(l);
126     return l;
127 }
128
129 void pa_ioline_close(pa_ioline *l) {
130     pa_assert(l);
131     pa_assert(PA_REFCNT_VALUE(l) >= 1);
132
133     l->dead = 1;
134
135     if (l->io) {
136         pa_iochannel_free(l->io);
137         l->io = NULL;
138     }
139
140     if (l->defer_event) {
141         l->mainloop->defer_free(l->defer_event);
142         l->defer_event = NULL;
143     }
144
145     if (l->callback)
146         l->callback = NULL;
147 }
148
149 void pa_ioline_puts(pa_ioline *l, const char *c) {
150     size_t len;
151
152     pa_assert(l);
153     pa_assert(PA_REFCNT_VALUE(l) >= 1);
154     pa_assert(c);
155
156     if (l->dead)
157         return;
158
159     len = strlen(c);
160     if (len > BUFFER_LIMIT - l->wbuf_valid_length)
161         len = BUFFER_LIMIT - l->wbuf_valid_length;
162
163     if (len) {
164         pa_assert(l->wbuf_length >= l->wbuf_valid_length);
165
166         /* In case the allocated buffer is too small, enlarge it. */
167         if (l->wbuf_valid_length + len > l->wbuf_length) {
168             size_t n = l->wbuf_valid_length+len;
169             char *new = pa_xmalloc(n);
170             if (l->wbuf) {
171                 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
172                 pa_xfree(l->wbuf);
173             }
174             l->wbuf = new;
175             l->wbuf_length = n;
176             l->wbuf_index = 0;
177         } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
178
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);
181             l->wbuf_index = 0;
182         }
183
184         pa_assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
185
186         /* Append the new string */
187         memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
188         l->wbuf_valid_length += len;
189
190         l->mainloop->defer_enable(l->defer_event, 1);
191     }
192 }
193
194 void pa_ioline_set_callback(pa_ioline*l, void (*callback)(pa_ioline*io, const char *s, void *userdata), void *userdata) {
195     pa_assert(l);
196     pa_assert(PA_REFCNT_VALUE(l) >= 1);
197
198     l->callback = callback;
199     l->userdata = userdata;
200 }
201
202 static void failure(pa_ioline *l, int process_leftover) {
203     pa_assert(l);
204     pa_assert(PA_REFCNT_VALUE(l) >= 1);
205     pa_assert(!l->dead);
206
207     if (process_leftover && l->rbuf_valid_length > 0) {
208         /* Pass the last missing bit to the client */
209
210         if (l->callback) {
211             char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
212             l->callback(l, p, l->userdata);
213             pa_xfree(p);
214         }
215     }
216
217     if (l->callback) {
218         l->callback(l, NULL, l->userdata);
219         l->callback = NULL;
220     }
221
222     pa_ioline_close(l);
223 }
224
225 static void scan_for_lines(pa_ioline *l, size_t skip) {
226     pa_assert(l);
227     pa_assert(PA_REFCNT_VALUE(l) >= 1);
228     pa_assert(skip < l->rbuf_valid_length);
229
230     while (!l->dead && l->rbuf_valid_length > skip) {
231         char *e, *p;
232         size_t m;
233
234         if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
235             break;
236
237         *e = 0;
238
239         p = l->rbuf + l->rbuf_index;
240         m = strlen(p);
241
242         l->rbuf_index += m+1;
243         l->rbuf_valid_length -= m+1;
244
245         /* A shortcut for the next time */
246         if (l->rbuf_valid_length == 0)
247             l->rbuf_index = 0;
248
249         if (l->callback)
250             l->callback(l, p, l->userdata);
251
252         skip = 0;
253     }
254
255     /* If the buffer became too large and still no newline was found, drop it. */
256     if (l->rbuf_valid_length >= BUFFER_LIMIT)
257         l->rbuf_index = l->rbuf_valid_length = 0;
258 }
259
260 static int do_write(pa_ioline *l);
261
262 static int do_read(pa_ioline *l) {
263     pa_assert(l);
264     pa_assert(PA_REFCNT_VALUE(l) >= 1);
265
266     while (!l->dead && pa_iochannel_is_readable(l->io)) {
267         ssize_t r;
268         size_t len;
269
270         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
271
272         /* Check if we have to enlarge the read buffer */
273         if (len < READ_SIZE) {
274             size_t n = l->rbuf_valid_length+READ_SIZE;
275
276             if (n >= BUFFER_LIMIT)
277                 n = BUFFER_LIMIT;
278
279             if (l->rbuf_length >= n) {
280                 /* The current buffer is large enough, let's just move the data to the front */
281                 if (l->rbuf_valid_length)
282                     memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
283             } else {
284                 /* Enlarge the buffer */
285                 char *new = pa_xmalloc(n);
286                 if (l->rbuf_valid_length)
287                     memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
288                 pa_xfree(l->rbuf);
289                 l->rbuf = new;
290                 l->rbuf_length = n;
291             }
292
293             l->rbuf_index = 0;
294         }
295
296         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
297
298         pa_assert(len >= READ_SIZE);
299
300         /* Read some data */
301         if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
302
303             if (r < 0 && errno == EAGAIN)
304                 return 0;
305
306             if (r < 0 && errno != ECONNRESET) {
307                 pa_log("read(): %s", pa_cstrerror(errno));
308                 failure(l, 0);
309             } else
310                 failure(l, 1);
311
312             return -1;
313         }
314
315         l->rbuf_valid_length += r;
316
317         /* Look if a line has been terminated in the newly read data */
318         scan_for_lines(l, l->rbuf_valid_length - r);
319     }
320
321     return 0;
322 }
323
324 /* Try to flush the buffer */
325 static int do_write(pa_ioline *l) {
326     ssize_t r;
327
328     pa_assert(l);
329     pa_assert(PA_REFCNT_VALUE(l) >= 1);
330
331     while (!l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length) {
332
333         if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
334
335             if (r < 0 && errno == EAGAIN)
336                 return 0;
337
338             if (r < 0 && errno != EPIPE)
339                 pa_log("write(): %s", pa_cstrerror(errno));
340
341             failure(l, 0);
342
343             return -1;
344         }
345
346         l->wbuf_index += r;
347         l->wbuf_valid_length -= r;
348
349         /* A shortcut for the next time */
350         if (l->wbuf_valid_length == 0)
351             l->wbuf_index = 0;
352     }
353
354     return 0;
355 }
356
357 /* Try to flush read/write data */
358 static void do_work(pa_ioline *l) {
359     pa_assert(l);
360     pa_assert(PA_REFCNT_VALUE(l) >= 1);
361
362     pa_ioline_ref(l);
363
364     l->mainloop->defer_enable(l->defer_event, 0);
365
366     if (!l->dead)
367         do_read(l);
368
369     if (!l->dead)
370         do_write(l);
371
372     if (l->defer_close && !l->wbuf_valid_length)
373         failure(l, 1);
374
375     pa_ioline_unref(l);
376 }
377
378 static void io_callback(pa_iochannel*io, void *userdata) {
379     pa_ioline *l = userdata;
380
381     pa_assert(io);
382     pa_assert(l);
383     pa_assert(PA_REFCNT_VALUE(l) >= 1);
384
385     do_work(l);
386 }
387
388 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
389     pa_ioline *l = userdata;
390
391     pa_assert(l);
392     pa_assert(PA_REFCNT_VALUE(l) >= 1);
393     pa_assert(l->mainloop == m);
394     pa_assert(l->defer_event == e);
395
396     do_work(l);
397 }
398
399 void pa_ioline_defer_close(pa_ioline *l) {
400     pa_assert(l);
401     pa_assert(PA_REFCNT_VALUE(l) >= 1);
402
403     l->defer_close = 1;
404
405     if (!l->wbuf_valid_length)
406         l->mainloop->defer_enable(l->defer_event, 1);
407 }
408
409 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
410     char *t;
411     va_list ap;
412
413     pa_assert(l);
414     pa_assert(PA_REFCNT_VALUE(l) >= 1);
415
416     va_start(ap, format);
417     t = pa_vsprintf_malloc(format, ap);
418     va_end(ap);
419
420     pa_ioline_puts(l, t);
421     pa_xfree(t);
422 }