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