src/protocol-simple.c

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