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