src/pulsecore/ioline.c

   1 /***
   2   This file is part of PulseAudio.
   3
   4   Copyright 2004-2006 Lennart Poettering
   5
   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.1 of the License,
   9   or (at your option) any later version.
  10
  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.
  15
  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
  19   USA.
  20 ***/
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include <errno.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30
  31 #include <pulse/xmalloc.h>
  32
  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>
  38
  39 #include "ioline.h"
  40
  41 #define BUFFER_LIMIT (64*1024)
  42 #define READ_SIZE (1024)
  43
  44 struct pa_ioline {
  45     PA_REFCNT_DECLARE;
  46
  47     pa_iochannel *io;
  48     pa_defer_event *defer_event;
  49     pa_mainloop_api *mainloop;
  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     pa_ioline_cb_t callback;
  58     void *userdata;
  59
  60     pa_ioline_drain_cb_t drain_callback;
  61     void *drain_userdata;
  62
  63     pa_bool_t dead:1;
  64     pa_bool_t defer_close:1;
  65 };
  66
  67 static void io_callback(pa_iochannel*io, void *userdata);
  68 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata);
  69
  70 pa_ioline* pa_ioline_new(pa_iochannel *io) {
  71     pa_ioline *l;
  72     pa_assert(io);
  73
  74     l = pa_xnew(pa_ioline, 1);
  75     PA_REFCNT_INIT(l);
  76     l->io = io;
  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->drain_callback = NULL;
  88     l->drain_userdata = NULL;
  89
  90     l->mainloop = pa_iochannel_get_mainloop_api(io);
  91
  92     l->defer_event = l->mainloop->defer_new(l->mainloop, defer_callback, l);
  93     l->mainloop->defer_enable(l->defer_event, 0);
  94
  95     l->dead = FALSE;
  96     l->defer_close = FALSE;
  97
  98     pa_iochannel_set_callback(io, io_callback, l);
  99
 100     return l;
 101 }
 102
 103 static void ioline_free(pa_ioline *l) {
 104     pa_assert(l);
 105
 106     if (l->io)
 107         pa_iochannel_free(l->io);
 108
 109     if (l->defer_event)
 110         l->mainloop->defer_free(l->defer_event);
 111
 112     pa_xfree(l->wbuf);
 113     pa_xfree(l->rbuf);
 114     pa_xfree(l);
 115 }
 116
 117 void pa_ioline_unref(pa_ioline *l) {
 118     pa_assert(l);
 119     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 120
 121     if (PA_REFCNT_DEC(l) <= 0)
 122         ioline_free(l);
 123 }
 124
 125 pa_ioline* pa_ioline_ref(pa_ioline *l) {
 126     pa_assert(l);
 127     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 128
 129     PA_REFCNT_INC(l);
 130     return l;
 131 }
 132
 133 void pa_ioline_close(pa_ioline *l) {
 134     pa_assert(l);
 135     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 136
 137     l->dead = TRUE;
 138
 139     if (l->io) {
 140         pa_iochannel_free(l->io);
 141         l->io = NULL;
 142     }
 143
 144     if (l->defer_event) {
 145         l->mainloop->defer_free(l->defer_event);
 146         l->defer_event = NULL;
 147     }
 148
 149     if (l->callback)
 150         l->callback = NULL;
 151 }
 152
 153 void pa_ioline_puts(pa_ioline *l, const char *c) {
 154     size_t len;
 155
 156     pa_assert(l);
 157     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 158     pa_assert(c);
 159
 160     if (l->dead)
 161         return;
 162
 163     len = strlen(c);
 164     if (len > BUFFER_LIMIT - l->wbuf_valid_length)
 165         len = BUFFER_LIMIT - l->wbuf_valid_length;
 166
 167     if (len) {
 168         pa_assert(l->wbuf_length >= l->wbuf_valid_length);
 169
 170         /* In case the allocated buffer is too small, enlarge it. */
 171         if (l->wbuf_valid_length + len > l->wbuf_length) {
 172             size_t n = l->wbuf_valid_length+len;
 173             char *new = pa_xnew(char, (unsigned) n);
 174
 175             if (l->wbuf) {
 176                 memcpy(new, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 177                 pa_xfree(l->wbuf);
 178             }
 179
 180             l->wbuf = new;
 181             l->wbuf_length = n;
 182             l->wbuf_index = 0;
 183         } else if (l->wbuf_index + l->wbuf_valid_length + len > l->wbuf_length) {
 184
 185             /* In case the allocated buffer fits, but the current index is too far from the start, move it to the front. */
 186             memmove(l->wbuf, l->wbuf+l->wbuf_index, l->wbuf_valid_length);
 187             l->wbuf_index = 0;
 188         }
 189
 190         pa_assert(l->wbuf_index + l->wbuf_valid_length + len <= l->wbuf_length);
 191
 192         /* Append the new string */
 193         memcpy(l->wbuf + l->wbuf_index + l->wbuf_valid_length, c, len);
 194         l->wbuf_valid_length += len;
 195
 196         l->mainloop->defer_enable(l->defer_event, 1);
 197     }
 198 }
 199
 200 void pa_ioline_set_callback(pa_ioline*l, pa_ioline_cb_t callback, void *userdata) {
 201     pa_assert(l);
 202     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 203
 204     if (l->dead)
 205         return;
 206
 207     l->callback = callback;
 208     l->userdata = userdata;
 209 }
 210
 211 void pa_ioline_set_drain_callback(pa_ioline*l, pa_ioline_drain_cb_t callback, void *userdata) {
 212     pa_assert(l);
 213     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 214
 215     if (l->dead)
 216         return;
 217
 218     l->drain_callback = callback;
 219     l->drain_userdata = userdata;
 220 }
 221
 222 static void failure(pa_ioline *l, pa_bool_t process_leftover) {
 223     pa_assert(l);
 224     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 225     pa_assert(!l->dead);
 226
 227     if (process_leftover && l->rbuf_valid_length > 0) {
 228         /* Pass the last missing bit to the client */
 229
 230         if (l->callback) {
 231             char *p = pa_xstrndup(l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 232             l->callback(l, p, l->userdata);
 233             pa_xfree(p);
 234         }
 235     }
 236
 237     if (l->callback) {
 238         l->callback(l, NULL, l->userdata);
 239         l->callback = NULL;
 240     }
 241
 242     pa_ioline_close(l);
 243 }
 244
 245 static void scan_for_lines(pa_ioline *l, size_t skip) {
 246     pa_assert(l);
 247     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 248     pa_assert(skip < l->rbuf_valid_length);
 249
 250     while (!l->dead && l->rbuf_valid_length > skip) {
 251         char *e, *p;
 252         size_t m;
 253
 254         if (!(e = memchr(l->rbuf + l->rbuf_index + skip, '\n', l->rbuf_valid_length - skip)))
 255             break;
 256
 257         *e = 0;
 258
 259         p = l->rbuf + l->rbuf_index;
 260         m = strlen(p);
 261
 262         l->rbuf_index += m+1;
 263         l->rbuf_valid_length -= m+1;
 264
 265         /* A shortcut for the next time */
 266         if (l->rbuf_valid_length == 0)
 267             l->rbuf_index = 0;
 268
 269         if (l->callback)
 270             l->callback(l, pa_strip_nl(p), l->userdata);
 271
 272         skip = 0;
 273     }
 274
 275     /* If the buffer became too large and still no newline was found, drop it. */
 276     if (l->rbuf_valid_length >= BUFFER_LIMIT)
 277         l->rbuf_index = l->rbuf_valid_length = 0;
 278 }
 279
 280 static int do_write(pa_ioline *l);
 281
 282 static int do_read(pa_ioline *l) {
 283     pa_assert(l);
 284     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 285
 286     while (l->io && !l->dead && pa_iochannel_is_readable(l->io)) {
 287         ssize_t r;
 288         size_t len;
 289
 290         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 291
 292         /* Check if we have to enlarge the read buffer */
 293         if (len < READ_SIZE) {
 294             size_t n = l->rbuf_valid_length+READ_SIZE;
 295
 296             if (n >= BUFFER_LIMIT)
 297                 n = BUFFER_LIMIT;
 298
 299             if (l->rbuf_length >= n) {
 300                 /* The current buffer is large enough, let's just move the data to the front */
 301                 if (l->rbuf_valid_length)
 302                     memmove(l->rbuf, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 303             } else {
 304                 /* Enlarge the buffer */
 305                 char *new = pa_xnew(char, (unsigned) n);
 306                 if (l->rbuf_valid_length)
 307                     memcpy(new, l->rbuf+l->rbuf_index, l->rbuf_valid_length);
 308                 pa_xfree(l->rbuf);
 309                 l->rbuf = new;
 310                 l->rbuf_length = n;
 311             }
 312
 313             l->rbuf_index = 0;
 314         }
 315
 316         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 317
 318         pa_assert(len >= READ_SIZE);
 319
 320         /* Read some data */
 321         if ((r = pa_iochannel_read(l->io, l->rbuf+l->rbuf_index+l->rbuf_valid_length, len)) <= 0) {
 322
 323             if (r < 0 && errno == EAGAIN)
 324                 return 0;
 325
 326             if (r < 0 && errno != ECONNRESET) {
 327                 pa_log("read(): %s", pa_cstrerror(errno));
 328                 failure(l, FALSE);
 329             } else
 330                 failure(l, TRUE);
 331
 332             return -1;
 333         }
 334
 335         l->rbuf_valid_length += (size_t) r;
 336
 337         /* Look if a line has been terminated in the newly read data */
 338         scan_for_lines(l, l->rbuf_valid_length - (size_t) r);
 339     }
 340
 341     return 0;
 342 }
 343
 344 /* Try to flush the buffer */
 345 static int do_write(pa_ioline *l) {
 346     ssize_t r;
 347
 348     pa_assert(l);
 349     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 350
 351     while (l->io && !l->dead && pa_iochannel_is_writable(l->io) && l->wbuf_valid_length > 0) {
 352
 353         if ((r = pa_iochannel_write(l->io, l->wbuf+l->wbuf_index, l->wbuf_valid_length)) <= 0) {
 354
 355             if (r < 0 && errno == EAGAIN)
 356                 break;
 357
 358             if (r < 0 && errno != EPIPE)
 359                 pa_log("write(): %s", pa_cstrerror(errno));
 360
 361             failure(l, FALSE);
 362
 363             return -1;
 364         }
 365
 366         l->wbuf_index += (size_t) r;
 367         l->wbuf_valid_length -= (size_t) r;
 368
 369         /* A shortcut for the next time */
 370         if (l->wbuf_valid_length == 0)
 371             l->wbuf_index = 0;
 372     }
 373
 374     if (l->wbuf_valid_length <= 0 && l->drain_callback)
 375         l->drain_callback(l, l->drain_userdata);
 376
 377     return 0;
 378 }
 379
 380 /* Try to flush read/write data */
 381 static void do_work(pa_ioline *l) {
 382     pa_assert(l);
 383     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 384
 385     pa_ioline_ref(l);
 386
 387     l->mainloop->defer_enable(l->defer_event, 0);
 388
 389     if (!l->dead)
 390         do_read(l);
 391
 392     if (!l->dead)
 393         do_write(l);
 394
 395     if (l->defer_close && !l->wbuf_valid_length)
 396         failure(l, TRUE);
 397
 398     pa_ioline_unref(l);
 399 }
 400
 401 static void io_callback(pa_iochannel*io, void *userdata) {
 402     pa_ioline *l = userdata;
 403
 404     pa_assert(io);
 405     pa_assert(l);
 406     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 407
 408     do_work(l);
 409 }
 410
 411 static void defer_callback(pa_mainloop_api*m, pa_defer_event*e, void *userdata) {
 412     pa_ioline *l = userdata;
 413
 414     pa_assert(l);
 415     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 416     pa_assert(l->mainloop == m);
 417     pa_assert(l->defer_event == e);
 418
 419     do_work(l);
 420 }
 421
 422 void pa_ioline_defer_close(pa_ioline *l) {
 423     pa_assert(l);
 424     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 425
 426     l->defer_close = TRUE;
 427
 428     if (!l->wbuf_valid_length)
 429         l->mainloop->defer_enable(l->defer_event, 1);
 430 }
 431
 432 void pa_ioline_printf(pa_ioline *l, const char *format, ...) {
 433     char *t;
 434     va_list ap;
 435
 436     pa_assert(l);
 437     pa_assert(PA_REFCNT_VALUE(l) >= 1);
 438
 439     va_start(ap, format);
 440     t = pa_vsprintf_malloc(format, ap);
 441     va_end(ap);
 442
 443     pa_ioline_puts(l, t);
 444     pa_xfree(t);
 445 }
 446
 447 pa_iochannel* pa_ioline_detach_iochannel(pa_ioline *l) {
 448     pa_iochannel *r;
 449
 450     pa_assert(l);
 451
 452     if (!l->io)
 453         return NULL;
 454
 455     r = l->io;
 456     l->io = NULL;
 457
 458     pa_iochannel_set_callback(r, NULL, NULL);
 459
 460     return r;
 461 }
 462
 463 pa_bool_t pa_ioline_is_drained(pa_ioline *l) {
 464     pa_assert(l);
 465
 466     return l->wbuf_valid_length <= 0;
 467 }