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