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 <assert.h>
  31 #include <stdlib.h>
  32 #include <string.h>
  33
  34 #include <pulse/xmalloc.h>
  35
  36 #include <pulsecore/core-error.h>
  37 #include <pulsecore/log.h>
  38
  39 #include "ioline.h"
  40
  41 #define BUFFER_LIMIT (64*1024)
  42 #define READ_SIZE (1024)
  43
  44 struct pa_ioline {
  45     pa_iochannel *io;
  46     pa_defer_event *defer_event;
  47     pa_mainloop_api *mainloop;
  48     int ref;
  49     int dead;
  50
  51     char *wbuf;
  52     size_t wbuf_length, wbuf_index, wbuf_valid_length;
  53
  54     char *rbuf;
  55     size_t rbuf_length, rbuf_index, rbuf_valid_length;
  56
  57     void (*callback)(pa_ioline*io, const char *s, void *userdata);
  58     void *userdata;
  59
  60     int defer_close;
  61 };
  62
  63 static void io_callback(pa_iochannel*io, void *userdata);
  64 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata);
  65
  66 pa_ioline* pa_ioline_new(pa_iochannel *io) {
  67     pa_ioline *l;
  68     assert(io);
  69
  70     l = pa_xnew(pa_ioline, 1);
  71     l->io = io;
  72     l->dead = 0;
  73
  74     l->wbuf = NULL;
  75     l->wbuf_length = l->wbuf_index = l->wbuf_valid_length = 0;
  76
  77     l->rbuf = NULL;
  78     l->rbuf_length = l->rbuf_index = l->rbuf_valid_length = 0;
  79
  80     l->callback = NULL;
  81     l->userdata = NULL;
  82     l->ref = 1;
  83
  84     l->mainloop = pa_iochannel_get_mainloop_api(io);
  85
  86     l->defer_event = l->mainloop->defer_new(l->mainloop, defer_callback, l);
  87     l->mainloop->defer_enable(l->defer_event, 0);
  88
  89     l->defer_close = 0;
  90
  91     pa_iochannel_set_callback(io, io_callback, l);
  92
  93     return l;
  94 }
  95
  96 static void ioline_free(pa_ioline *l) {
  97     assert(l);
  98
  99     if (l->io)
 100         pa_iochannel_free(l->io);
 101
 102     if (l->defer_event)
 103         l->mainloop->defer_free(l->defer_event);
 104
 105     pa_xfree(l->wbuf);
 106     pa_xfree(l->rbuf);
 107     pa_xfree(l);
 108 }
 109
 110 void pa_ioline_unref(pa_ioline *l) {
 111     assert(l);
 112     assert(l->ref >= 1);
 113
 114     if ((--l->ref) <= 0)
 115         ioline_free(l);
 116 }
 117
 118 pa_ioline* pa_ioline_ref(pa_ioline *l) {
 119     assert(l);
 120     assert(l->ref >= 1);
 121
 122     l->ref++;
 123     return l;
 124 }
 125
 126 void pa_ioline_close(pa_ioline *l) {
 127     assert(l);
 128     assert(l->ref >= 1);
 129
 130     l->dead = 1;
 131
 132     if (l->io) {
 133         pa_iochannel_free(l->io);
 134         l->io = NULL;
 135     }
 136
 137     if (l->defer_event) {
 138         l->mainloop->defer_free(l->defer_event);
 139         l->defer_event = NULL;
 140     }
 141
 142     if (l->callback)
 143         l->callback = NULL;
 144 }
 145
 146 void pa_ioline_puts(pa_ioline *l, const char *c) {
 147     size_t len;
 148
 149     assert(l);
 150     assert(l->ref >= 1);
 151     assert(c);
 152
 153     if (l->dead)
 154         return;
 155
 156     len = strlen(c);
 157     if (len > BUFFER_LIMIT - l->wbuf_valid_length)
 158         len = BUFFER_LIMIT - l->wbuf_valid_length;
 159
 160     if (len) {
 161         assert(l->wbuf_length >= l->wbuf_valid_length);
 162
 163         /* In case the allocated buffer is too small, enlarge it. */
 164         if (l->wbuf_valid_length + len > l->wbuf_length) {
 165             size_t n = l->wbuf_valid_length+len;
 166             char *new = pa_xmalloc(n);
 167             if (l->wbuf) {
 168                 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 169                 pa_xfree(l->wbuf);
 170             }
 171             l->wbuf = new;
 172             l->wbuf_length = n;
 173             l->wbuf_index = 0;
 174         } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
 175
 176             /* In case the allocated buffer fits, but the current index is too far from the start, move it to the front. */
 177             memmove(l->wbuf, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 178             l->wbuf_index = 0;
 179         }
 180
 181         assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
 182
 183         /* Append the new string */
 184         memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
 185         l->wbuf_valid_length += len;
 186
 187         l->mainloop->defer_enable(l->defer_event, 1);
 188     }
 189 }
 190
 191 void pa_ioline_set_callback(pa_ioline*l, void (*callback)(pa_ioline*io, const char *s, void *userdata), void *userdata) {
 192     assert(l);
 193     assert(l->ref >= 1);
 194
 195     l->callback = callback;
 196     l->userdata = userdata;
 197 }
 198
 199 static void failure(pa_ioline *l, int process_leftover) {
 200     assert(l);
 201     assert(l->ref >= 1);
 202     assert(!l->dead);
 203
 204     if (process_leftover && l->rbuf_valid_length > 0) {
 205         /* Pass the last missing bit to the client */
 206
 207         if (l->callback) {
 208             char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 209             l->callback(l, p, l->userdata);
 210             pa_xfree(p);
 211         }
 212     }
 213
 214     if (l->callback) {
 215         l->callback(l, NULL, l->userdata);
 216         l->callback = NULL;
 217     }
 218
 219     pa_ioline_close(l);
 220 }
 221
 222 static void scan_for_lines(pa_ioline *l, size_t skip) {
 223     assert(l && l->ref >= 1 && skip < l->rbuf_valid_length);
 224
 225     while (!l->dead && l->rbuf_valid_length > skip) {
 226         char *e, *p;
 227         size_t m;
 228
 229         if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
 230             break;
 231
 232         *e = 0;
 233
 234         p = l->rbuf + l->rbuf_index;
 235         m = strlen(p);
 236
 237         l->rbuf_index += m+1;
 238         l->rbuf_valid_length -= m+1;
 239
 240         /* A shortcut for the next time */
 241         if (l->rbuf_valid_length == 0)
 242             l->rbuf_index = 0;
 243
 244         if (l->callback)
 245             l->callback(l, p, l->userdata);
 246
 247         skip = 0;
 248     }
 249
 250     /* If the buffer became too large and still no newline was found, drop it. */
 251     if (l->rbuf_valid_length >= BUFFER_LIMIT)
 252         l->rbuf_index = l->rbuf_valid_length = 0;
 253 }
 254
 255 static int do_write(pa_ioline *l);
 256
 257 static int do_read(pa_ioline *l) {
 258     assert(l && l->ref >= 1);
 259
 260     while (!l->dead && pa_iochannel_is_readable(l->io)) {
 261         ssize_t r;
 262         size_t len;
 263
 264         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 265
 266         /* Check if we have to enlarge the read buffer */
 267         if (len < READ_SIZE) {
 268             size_t n = l->rbuf_valid_length+READ_SIZE;
 269
 270             if (n >= BUFFER_LIMIT)
 271                 n = BUFFER_LIMIT;
 272
 273             if (l->rbuf_length >= n) {
 274                 /* The current buffer is large enough, let's just move the data to the front */
 275                 if (l->rbuf_valid_length)
 276                     memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 277             } else {
 278                 /* Enlarge the buffer */
 279                 char *new = pa_xmalloc(n);
 280                 if (l->rbuf_valid_length)
 281                     memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 282                 pa_xfree(l->rbuf);
 283                 l->rbuf = new;
 284                 l->rbuf_length = n;
 285             }
 286
 287             l->rbuf_index = 0;
 288         }
 289
 290         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 291
 292         assert(len >= READ_SIZE);
 293
 294         /* Read some data */
 295         if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
 296             if (r < 0 && errno != ECONNRESET) {
 297                 pa_log("read(): %s", pa_cstrerror(errno));
 298                 failure(l, 0);
 299             } else
 300                 failure(l, 1);
 301
 302             return -1;
 303         }
 304
 305         l->rbuf_valid_length += r;
 306
 307         /* Look if a line has been terminated in the newly read data */
 308         scan_for_lines(l, l->rbuf_valid_length - r);
 309     }
 310
 311     return 0;
 312 }
 313
 314 /* Try to flush the buffer */
 315 static int do_write(pa_ioline *l) {
 316     ssize_t r;
 317     assert(l && l->ref >= 1);
 318
 319     while (!l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length) {
 320
 321         if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
 322
 323             if (r < 0 && errno != EPIPE)
 324                 pa_log("write(): %s", pa_cstrerror(errno));
 325
 326             failure(l, 0);
 327
 328             return -1;
 329         }
 330
 331         l->wbuf_index += r;
 332         l->wbuf_valid_length -= r;
 333
 334         /* A shortcut for the next time */
 335         if (l->wbuf_valid_length == 0)
 336             l->wbuf_index = 0;
 337     }
 338
 339     return 0;
 340 }
 341
 342 /* Try to flush read/write data */
 343 static void do_work(pa_ioline *l) {
 344     assert(l);
 345     assert(l->ref >= 1);
 346
 347     pa_ioline_ref(l);
 348
 349     l->mainloop->defer_enable(l->defer_event, 0);
 350
 351     if (!l->dead)
 352         do_read(l);
 353
 354     if (!l->dead)
 355         do_write(l);
 356
 357     if (l->defer_close && !l->wbuf_valid_length)
 358         failure(l, 1);
 359
 360     pa_ioline_unref(l);
 361 }
 362
 363 static void io_callback(pa_iochannel*io, void *userdata) {
 364     pa_ioline *l = userdata;
 365     assert(io && l && l->ref >= 1);
 366
 367     do_work(l);
 368 }
 369
 370 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
 371     pa_ioline *l = userdata;
 372     assert(l && l->ref >= 1 && l->mainloop == m && l->defer_event == e);
 373
 374     do_work(l);
 375 }
 376
 377 void pa_ioline_defer_close(pa_ioline *l) {
 378     assert(l);
 379     assert(l->ref >= 1);
 380
 381     l->defer_close = 1;
 382
 383     if (!l->wbuf_valid_length)
 384         l->mainloop->defer_enable(l->defer_event, 1);
 385 }
 386
 387 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
 388     char *t;
 389     va_list ap;
 390
 391     assert(l);
 392     assert(l->ref >= 1);
 393
 394     va_start(ap, format);
 395     t = pa_vsprintf_malloc(format, ap);
 396     va_end(ap);
 397
 398     pa_ioline_puts(l, t);
 399     pa_xfree(t);
 400 }