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