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
  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     pa_ioline_cb_t callback;
  60     void *userdata;
  61
  62     pa_bool_t dead:1;
  63     pa_bool_t defer_close:1;
  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
  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->dead = FALSE;
  92     l->defer_close = FALSE;
  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 = TRUE;
 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_xnew(char, n);
 170
 171             if (l->wbuf) {
 172                 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 173                 pa_xfree(l->wbuf);
 174             }
 175
 176             l->wbuf = new;
 177             l->wbuf_length = n;
 178             l->wbuf_index = 0;
 179         } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
 180
 181             /* In case the allocated buffer fits, but the current index is too far from the start, move it to the front. */
 182             memmove(l->wbuf, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 183             l->wbuf_index = 0;
 184         }
 185
 186         pa_assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
 187
 188         /* Append the new string */
 189         memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
 190         l->wbuf_valid_length += len;
 191
 192         l->mainloop->defer_enable(l->defer_event, 1);
 193     }
 194 }
 195
 196 void pa_ioline_set_callback(pa_ioline*l, pa_ioline_cb_t callback, void *userdata) {
 197     pa_assert(l);
 198     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 199
 200     if (l->dead)
 201         return;
 202
 203     l->callback = callback;
 204     l->userdata = userdata;
 205 }
 206
 207 static void failure(pa_ioline *l, pa_bool_t process_leftover) {
 208     pa_assert(l);
 209     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 210     pa_assert(!l->dead);
 211
 212     if (process_leftover && l->rbuf_valid_length > 0) {
 213         /* Pass the last missing bit to the client */
 214
 215         if (l->callback) {
 216             char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 217             l->callback(l, p, l->userdata);
 218             pa_xfree(p);
 219         }
 220     }
 221
 222     if (l->callback) {
 223         l->callback(l, NULL, l->userdata);
 224         l->callback = NULL;
 225     }
 226
 227     pa_ioline_close(l);
 228 }
 229
 230 static void scan_for_lines(pa_ioline *l, size_t skip) {
 231     pa_assert(l);
 232     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 233     pa_assert(skip < l->rbuf_valid_length);
 234
 235     while (!l->dead && l->rbuf_valid_length > skip) {
 236         char *e, *p;
 237         size_t m;
 238
 239         if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
 240             break;
 241
 242         *e = 0;
 243
 244         p = l->rbuf + l->rbuf_index;
 245         m = strlen(p);
 246
 247         l->rbuf_index += m+1;
 248         l->rbuf_valid_length -= m+1;
 249
 250         /* A shortcut for the next time */
 251         if (l->rbuf_valid_length == 0)
 252             l->rbuf_index = 0;
 253
 254         if (l->callback)
 255             l->callback(l, pa_strip_nl(p), l->userdata);
 256
 257         skip = 0;
 258     }
 259
 260     /* If the buffer became too large and still no newline was found, drop it. */
 261     if (l->rbuf_valid_length >= BUFFER_LIMIT)
 262         l->rbuf_index = l->rbuf_valid_length = 0;
 263 }
 264
 265 static int do_write(pa_ioline *l);
 266
 267 static int do_read(pa_ioline *l) {
 268     pa_assert(l);
 269     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 270
 271     while (!l->dead && pa_iochannel_is_readable(l->io)) {
 272         ssize_t r;
 273         size_t len;
 274
 275         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 276
 277         /* Check if we have to enlarge the read buffer */
 278         if (len < READ_SIZE) {
 279             size_t n = l->rbuf_valid_length+READ_SIZE;
 280
 281             if (n >= BUFFER_LIMIT)
 282                 n = BUFFER_LIMIT;
 283
 284             if (l->rbuf_length >= n) {
 285                 /* The current buffer is large enough, let's just move the data to the front */
 286                 if (l->rbuf_valid_length)
 287                     memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 288             } else {
 289                 /* Enlarge the buffer */
 290                 char *new = pa_xnew(char, n);
 291                 if (l->rbuf_valid_length)
 292                     memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 293                 pa_xfree(l->rbuf);
 294                 l->rbuf = new;
 295                 l->rbuf_length = n;
 296             }
 297
 298             l->rbuf_index = 0;
 299         }
 300
 301         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 302
 303         pa_assert(len >= READ_SIZE);
 304
 305         /* Read some data */
 306         if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
 307
 308             if (r < 0 && errno == EAGAIN)
 309                 return 0;
 310
 311             if (r < 0 && errno != ECONNRESET) {
 312                 pa_log("read(): %s", pa_cstrerror(errno));
 313                 failure(l, FALSE);
 314             } else
 315                 failure(l, TRUE);
 316
 317             return -1;
 318         }
 319
 320         l->rbuf_valid_length += r;
 321
 322         /* Look if a line has been terminated in the newly read data */
 323         scan_for_lines(l, l->rbuf_valid_length - r);
 324     }
 325
 326     return 0;
 327 }
 328
 329 /* Try to flush the buffer */
 330 static int do_write(pa_ioline *l) {
 331     ssize_t r;
 332
 333     pa_assert(l);
 334     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 335
 336     while (!l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length) {
 337
 338         if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
 339
 340             if (r < 0 && errno == EAGAIN)
 341                 return 0;
 342
 343             if (r < 0 && errno != EPIPE)
 344                 pa_log("write(): %s", pa_cstrerror(errno));
 345
 346             failure(l, FALSE);
 347
 348             return -1;
 349         }
 350
 351         l->wbuf_index += r;
 352         l->wbuf_valid_length -= r;
 353
 354         /* A shortcut for the next time */
 355         if (l->wbuf_valid_length == 0)
 356             l->wbuf_index = 0;
 357     }
 358
 359     return 0;
 360 }
 361
 362 /* Try to flush read/write data */
 363 static void do_work(pa_ioline *l) {
 364     pa_assert(l);
 365     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 366
 367     pa_ioline_ref(l);
 368
 369     l->mainloop->defer_enable(l->defer_event, 0);
 370
 371     if (!l->dead)
 372         do_read(l);
 373
 374     if (!l->dead)
 375         do_write(l);
 376
 377     if (l->defer_close && !l->wbuf_valid_length)
 378         failure(l, TRUE);
 379
 380     pa_ioline_unref(l);
 381 }
 382
 383 static void io_callback(pa_iochannel*io, void *userdata) {
 384     pa_ioline *l = userdata;
 385
 386     pa_assert(io);
 387     pa_assert(l);
 388     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 389
 390     do_work(l);
 391 }
 392
 393 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
 394     pa_ioline *l = userdata;
 395
 396     pa_assert(l);
 397     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 398     pa_assert(l->mainloop == m);
 399     pa_assert(l->defer_event == e);
 400
 401     do_work(l);
 402 }
 403
 404 void pa_ioline_defer_close(pa_ioline *l) {
 405     pa_assert(l);
 406     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 407
 408     l->defer_close = TRUE;
 409
 410     if (!l->wbuf_valid_length)
 411         l->mainloop->defer_enable(l->defer_event, 1);
 412 }
 413
 414 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
 415     char *t;
 416     va_list ap;
 417
 418     pa_assert(l);
 419     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 420
 421     va_start(ap, format);
 422     t = pa_vsprintf_malloc(format, ap);
 423     va_end(ap);
 424
 425     pa_ioline_puts(l, t);
 426     pa_xfree(t);
 427 }