code.delx.au - pulseaudio/blob - polyp/protocol-simple.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of polypaudio.
   5
   6   polypaudio 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   polypaudio 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 polypaudio; 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 <assert.h>
  27 #include <stdlib.h>
  28 #include <limits.h>
  29 #include <stdio.h>
  30 #include <errno.h>
  31 #include <string.h>
  32
  33 #include "sink-input.h"
  34 #include "source-output.h"
  35 #include "protocol-simple.h"
  36 #include "client.h"
  37 #include "sample-util.h"
  38 #include "namereg.h"
  39 #include "xmalloc.h"
  40 #include "log.h"
  41
  42 /* Don't allow more than this many concurrent connections */
  43 #define MAX_CONNECTIONS 10
  44
  45 #define PA_TYPEID_SIMPLE PA_TYPEID_MAKE('S', 'M', 'P', 'L')
  46
  47 struct connection {
  48     struct pa_protocol_simple *protocol;
  49     struct pa_iochannel *io;
  50     struct pa_sink_input *sink_input;
  51     struct pa_source_output *source_output;
  52     struct pa_client *client;
  53     struct pa_memblockq *input_memblockq, *output_memblockq;
  54     struct pa_defer_event *defer_event;
  55
  56     struct {
  57         struct pa_memblock *current_memblock;
  58         size_t memblock_index, fragment_size;
  59     } playback;
  60 };
  61
  62 struct pa_protocol_simple {
  63     struct pa_module *module;
  64     struct pa_core *core;
  65     struct pa_socket_server*server;
  66     struct pa_idxset *connections;
  67     enum {
  68         RECORD = 1,
  69         PLAYBACK = 2,
  70         DUPLEX = 3
  71     } mode;
  72     struct pa_sample_spec sample_spec;
  73     char *source_name, *sink_name;
  74 };
  75
  76 #define PLAYBACK_BUFFER_SECONDS (.5)
  77 #define PLAYBACK_BUFFER_FRAGMENTS (10)
  78 #define RECORD_BUFFER_SECONDS (5)
  79 #define RECORD_BUFFER_FRAGMENTS (100)
  80
  81 static void connection_free(struct connection *c) {
  82     assert(c);
  83
  84     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
  85
  86     if (c->playback.current_memblock)
  87         pa_memblock_unref(c->playback.current_memblock);
  88     if (c->sink_input) {
  89         pa_sink_input_disconnect(c->sink_input);
  90         pa_sink_input_unref(c->sink_input);
  91     }
  92     if (c->source_output) {
  93         pa_source_output_disconnect(c->source_output);
  94         pa_source_output_unref(c->source_output);
  95     }
  96     if (c->client)
  97         pa_client_free(c->client);
  98     if (c->io)
  99         pa_iochannel_free(c->io);
 100     if (c->input_memblockq)
 101         pa_memblockq_free(c->input_memblockq);
 102     if (c->output_memblockq)
 103         pa_memblockq_free(c->output_memblockq);
 104     if (c->defer_event)
 105         c->protocol->core->mainloop->defer_free(c->defer_event);
 106     pa_xfree(c);
 107 }
 108
 109 static int do_read(struct connection *c) {
 110     struct pa_memchunk chunk;
 111     ssize_t r;
 112     size_t l;
 113
 114     if (!c->sink_input || !(l = pa_memblockq_missing(c->input_memblockq)))
 115         return 0;
 116
 117     if (l > c->playback.fragment_size)
 118         l = c->playback.fragment_size;
 119
 120     if (c->playback.current_memblock)
 121         if (c->playback.current_memblock->length - c->playback.memblock_index < l) {
 122             pa_memblock_unref(c->playback.current_memblock);
 123             c->playback.current_memblock = NULL;
 124             c->playback.memblock_index = 0;
 125         }
 126
 127     if (!c->playback.current_memblock) {
 128         c->playback.current_memblock = pa_memblock_new(c->playback.fragment_size*2, c->protocol->core->memblock_stat);
 129         assert(c->playback.current_memblock && c->playback.current_memblock->length >= l);
 130         c->playback.memblock_index = 0;
 131     }
 132
 133     if ((r = pa_iochannel_read(c->io, (uint8_t*) c->playback.current_memblock->data+c->playback.memblock_index, l)) <= 0) {
 134         pa_log(__FILE__": read() failed: %s\n", r == 0 ? "EOF" : strerror(errno));
 135         return -1;
 136     }
 137
 138     chunk.memblock = c->playback.current_memblock;
 139     chunk.index = c->playback.memblock_index;
 140     chunk.length = r;
 141     assert(chunk.memblock);
 142
 143     c->playback.memblock_index += r;
 144
 145     assert(c->input_memblockq);
 146     pa_memblockq_push_align(c->input_memblockq, &chunk, 0);
 147     assert(c->sink_input);
 148     pa_sink_notify(c->sink_input->sink);
 149
 150     return 0;
 151 }
 152
 153 static int do_write(struct connection *c) {
 154     struct pa_memchunk chunk;
 155     ssize_t r;
 156
 157     if (!c->source_output)
 158         return 0;
 159
 160     assert(c->output_memblockq);
 161     if (pa_memblockq_peek(c->output_memblockq, &chunk) < 0)
 162         return 0;
 163
 164     assert(chunk.memblock && chunk.length);
 165
 166     if ((r = pa_iochannel_write(c->io, (uint8_t*) chunk.memblock->data+chunk.index, chunk.length)) < 0) {
 167         pa_memblock_unref(chunk.memblock);
 168         pa_log(__FILE__": write(): %s\n", strerror(errno));
 169         return -1;
 170     }
 171
 172     pa_memblockq_drop(c->output_memblockq, &chunk, r);
 173     pa_memblock_unref(chunk.memblock);
 174
 175     return 0;
 176 }
 177
 178
 179 static void do_work(struct connection *c) {
 180     assert(c);
 181
 182     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable);
 183     c->protocol->core->mainloop->defer_enable(c->defer_event, 0);
 184
 185     if (pa_iochannel_is_writable(c->io))
 186         if (do_write(c) < 0)
 187             goto fail;
 188
 189     if (pa_iochannel_is_readable(c->io))
 190         if (do_read(c) < 0)
 191             goto fail;
 192
 193     if (pa_iochannel_is_hungup(c->io))
 194         c->protocol->core->mainloop->defer_enable(c->defer_event, 1);
 195
 196     return;
 197
 198 fail:
 199     connection_free(c);
 200 }
 201
 202 /*** sink_input callbacks ***/
 203
 204 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk) {
 205     struct connection*c;
 206     assert(i && i->userdata && chunk);
 207     c = i->userdata;
 208
 209     if (pa_memblockq_peek(c->input_memblockq, chunk) < 0)
 210         return -1;
 211
 212     return 0;
 213 }
 214
 215 static void sink_input_drop_cb(struct pa_sink_input *i, const struct pa_memchunk *chunk, size_t length) {
 216     struct connection*c = i->userdata;
 217     assert(i && c && length);
 218
 219     pa_memblockq_drop(c->input_memblockq, chunk, length);
 220
 221     /* do something */
 222     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable);
 223     c->protocol->core->mainloop->defer_enable(c->defer_event, 1);
 224 }
 225
 226 static void sink_input_kill_cb(struct pa_sink_input *i) {
 227     assert(i && i->userdata);
 228     connection_free((struct connection *) i->userdata);
 229 }
 230
 231
 232 static pa_usec_t sink_input_get_latency_cb(struct pa_sink_input *i) {
 233     struct connection*c = i->userdata;
 234     assert(i && c);
 235     return pa_bytes_to_usec(pa_memblockq_get_length(c->input_memblockq), &c->sink_input->sample_spec);
 236 }
 237
 238 /*** source_output callbacks ***/
 239
 240 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk) {
 241     struct connection *c = o->userdata;
 242     assert(o && c && chunk);
 243
 244     pa_memblockq_push(c->output_memblockq, chunk, 0);
 245
 246     /* do something */
 247     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop && c->protocol->core->mainloop->defer_enable);
 248     c->protocol->core->mainloop->defer_enable(c->defer_event, 1);
 249 }
 250
 251 static void source_output_kill_cb(struct pa_source_output *o) {
 252     assert(o && o->userdata);
 253     connection_free((struct connection *) o->userdata);
 254 }
 255
 256 static pa_usec_t source_output_get_latency_cb(struct pa_source_output *o) {
 257     struct connection*c = o->userdata;
 258     assert(o && c);
 259     return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
 260 }
 261
 262 /*** client callbacks ***/
 263
 264 static void client_kill_cb(struct pa_client *c) {
 265     assert(c && c->userdata);
 266     connection_free((struct connection *) c->userdata);
 267 }
 268
 269 /*** pa_iochannel callbacks ***/
 270
 271 static void io_callback(struct pa_iochannel*io, void *userdata) {
 272     struct connection *c = userdata;
 273     assert(io && c && c->io == io);
 274
 275     do_work(c);
 276 }
 277
 278 /*** fixed callback ***/
 279
 280 static void defer_callback(struct pa_mainloop_api*a, struct pa_defer_event *e, void *userdata) {
 281     struct connection *c = userdata;
 282     assert(a && c && c->defer_event == e);
 283
 284     do_work(c);
 285 }
 286
 287 /*** socket_server callbacks ***/
 288
 289 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
 290     struct pa_protocol_simple *p = userdata;
 291     struct connection *c = NULL;
 292     char cname[256];
 293     assert(s && io && p);
 294
 295     if (pa_idxset_ncontents(p->connections)+1 > MAX_CONNECTIONS) {
 296         pa_log(__FILE__": Warning! Too many connections (%u), dropping incoming connection.\n", MAX_CONNECTIONS);
 297         pa_iochannel_free(io);
 298         return;
 299     }
 300
 301     c = pa_xmalloc(sizeof(struct connection));
 302     c->io = io;
 303     c->sink_input = NULL;
 304     c->source_output = NULL;
 305     c->defer_event = NULL;
 306     c->input_memblockq = c->output_memblockq = NULL;
 307     c->protocol = p;
 308     c->playback.current_memblock = NULL;
 309     c->playback.memblock_index = 0;
 310     c->playback.fragment_size = 0;
 311
 312     pa_iochannel_socket_peer_to_string(io, cname, sizeof(cname));
 313     c->client = pa_client_new(p->core, PA_TYPEID_SIMPLE, cname);
 314     assert(c->client);
 315     c->client->owner = p->module;
 316     c->client->kill = client_kill_cb;
 317     c->client->userdata = c;
 318
 319     if (p->mode & PLAYBACK) {
 320         struct pa_sink *sink;
 321         size_t l;
 322
 323         if (!(sink = pa_namereg_get(p->core, p->sink_name, PA_NAMEREG_SINK, 1))) {
 324             pa_log(__FILE__": Failed to get sink.\n");
 325             goto fail;
 326         }
 327
 328         if (!(c->sink_input = pa_sink_input_new(sink, PA_TYPEID_SIMPLE, c->client->name, &p->sample_spec, 0, -1))) {
 329             pa_log(__FILE__": Failed to create sink input.\n");
 330             goto fail;
 331         }
 332
 333         c->sink_input->owner = p->module;
 334         c->sink_input->client = c->client;
 335
 336         c->sink_input->peek = sink_input_peek_cb;
 337         c->sink_input->drop = sink_input_drop_cb;
 338         c->sink_input->kill = sink_input_kill_cb;
 339         c->sink_input->get_latency = sink_input_get_latency_cb;
 340         c->sink_input->userdata = c;
 341
 342         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*PLAYBACK_BUFFER_SECONDS);
 343         c->input_memblockq = pa_memblockq_new(l, 0, pa_frame_size(&p->sample_spec), l/2, l/PLAYBACK_BUFFER_FRAGMENTS, p->core->memblock_stat);
 344         assert(c->input_memblockq);
 345         pa_iochannel_socket_set_rcvbuf(io, l/PLAYBACK_BUFFER_FRAGMENTS*5);
 346         c->playback.fragment_size = l/10;
 347     }
 348
 349     if (p->mode & RECORD) {
 350         struct pa_source *source;
 351         size_t l;
 352
 353         if (!(source = pa_namereg_get(p->core, p->source_name, PA_NAMEREG_SOURCE, 1))) {
 354             pa_log(__FILE__": Failed to get source.\n");
 355             goto fail;
 356         }
 357
 358         c->source_output = pa_source_output_new(source, PA_TYPEID_SIMPLE, c->client->name, &p->sample_spec, -1);
 359         if (!c->source_output) {
 360             pa_log(__FILE__": Failed to create source output.\n");
 361             goto fail;
 362         }
 363         c->source_output->owner = p->module;
 364         c->source_output->client = c->client;
 365
 366         c->source_output->push = source_output_push_cb;
 367         c->source_output->kill = source_output_kill_cb;
 368         c->source_output->get_latency = source_output_get_latency_cb;
 369         c->source_output->userdata = c;
 370
 371         l = (size_t) (pa_bytes_per_second(&p->sample_spec)*RECORD_BUFFER_SECONDS);
 372         c->output_memblockq = pa_memblockq_new(l, 0, pa_frame_size(&p->sample_spec), 0, 0, p->core->memblock_stat);
 373         pa_iochannel_socket_set_sndbuf(io, l/RECORD_BUFFER_FRAGMENTS*2);
 374     }
 375
 376     pa_iochannel_set_callback(c->io, io_callback, c);
 377     pa_idxset_put(p->connections, c, NULL);
 378
 379     c->defer_event = p->core->mainloop->defer_new(p->core->mainloop, defer_callback, c);
 380     assert(c->defer_event);
 381     p->core->mainloop->defer_enable(c->defer_event, 0);
 382
 383     return;
 384
 385 fail:
 386     if (c)
 387         connection_free(c);
 388 }
 389
 390 struct pa_protocol_simple* pa_protocol_simple_new(struct pa_core *core, struct pa_socket_server *server, struct pa_module *m, struct pa_modargs *ma) {
 391     struct pa_protocol_simple* p = NULL;
 392     int enable;
 393     assert(core && server && ma);
 394
 395     p = pa_xmalloc0(sizeof(struct pa_protocol_simple));
 396     p->module = m;
 397     p->core = core;
 398     p->server = server;
 399     p->connections = pa_idxset_new(NULL, NULL);
 400
 401     p->sample_spec = core->default_sample_spec;
 402     if (pa_modargs_get_sample_spec(ma, &p->sample_spec) < 0) {
 403         pa_log(__FILE__": Failed to parse sample type specification.\n");
 404         goto fail;
 405     }
 406
 407     p->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
 408     p->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
 409
 410     enable = 0;
 411     if (pa_modargs_get_value_boolean(ma, "record", &enable) < 0) {
 412         pa_log(__FILE__": record= expects a numeric argument.\n");
 413         goto fail;
 414     }
 415     p->mode = enable ? RECORD : 0;
 416
 417     enable = 1;
 418     if (pa_modargs_get_value_boolean(ma, "playback", &enable) < 0) {
 419         pa_log(__FILE__": playback= expects a numeric argument.\n");
 420         goto fail;
 421     }
 422     p->mode |= enable ? PLAYBACK : 0;
 423
 424     if ((p->mode & (RECORD|PLAYBACK)) == 0) {
 425         pa_log(__FILE__": neither playback nor recording enabled for protocol.\n");
 426         goto fail;
 427     }
 428
 429     pa_socket_server_set_callback(p->server, on_connection, p);
 430
 431     return p;
 432
 433 fail:
 434     if (p)
 435         pa_protocol_simple_free(p);
 436     return NULL;
 437 }
 438
 439
 440 void pa_protocol_simple_free(struct pa_protocol_simple *p) {
 441     struct connection *c;
 442     assert(p);
 443
 444     if (p->connections) {
 445         while((c = pa_idxset_first(p->connections, NULL)))
 446             connection_free(c);
 447
 448         pa_idxset_free(p->connections, NULL, NULL);
 449     }
 450
 451     if (p->server)
 452         pa_socket_server_unref(p->server);
 453     pa_xfree(p);
 454 }
 455