code.delx.au - pulseaudio/blob - 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
  13 struct connection {
  14     struct protocol_simple *protocol;
  15     struct iochannel *io;
  16     struct sink_input *sink_input;
  17     struct source_output *source_output;
  18     struct client *client;
  19     struct memblockq *input_memblockq, *output_memblockq;
  20 };
  21
  22 struct protocol_simple {
  23     struct core *core;
  24     struct socket_server*server;
  25     struct idxset *connections;
  26     enum protocol_simple_mode mode;
  27 };
  28
  29 #define BUFSIZE PIPE_BUF
  30
  31 static void free_connection(void *data, void *userdata) {
  32     struct connection *c = data;
  33     assert(data);
  34
  35     if (c->sink_input)
  36         sink_input_free(c->sink_input);
  37     if (c->source_output)
  38         source_output_free(c->source_output);
  39     if (c->client)
  40         client_free(c->client);
  41     if (c->io)
  42         iochannel_free(c->io);
  43     if (c->input_memblockq)
  44         memblockq_free(c->input_memblockq);
  45     if (c->output_memblockq)
  46         memblockq_free(c->output_memblockq);
  47     free(c);
  48 }
  49
  50 static void destroy_connection(struct connection *c) {
  51     assert(c && c->protocol);
  52     idxset_remove_by_data(c->protocol->connections, c, NULL);
  53     free_connection(c, NULL);
  54 }
  55
  56 static int do_read(struct connection *c) {
  57     struct memchunk chunk;
  58     ssize_t r;
  59
  60     if (!iochannel_is_readable(c->io))
  61         return 0;
  62
  63     if (!c->sink_input || !memblockq_is_writable(c->input_memblockq, BUFSIZE))
  64         return 0;
  65
  66     chunk.memblock = memblock_new(BUFSIZE);
  67     assert(chunk.memblock);
  68
  69     if ((r = iochannel_read(c->io, chunk.memblock->data, BUFSIZE)) <= 0) {
  70         fprintf(stderr, "read(): %s\n", r == 0 ? "EOF" : strerror(errno));
  71         memblock_unref(chunk.memblock);
  72         return -1;
  73     }
  74
  75     chunk.memblock->length = r;
  76     chunk.length = r;
  77     chunk.index = 0;
  78
  79     assert(c->input_memblockq);
  80     memblockq_push(c->input_memblockq, &chunk, 0);
  81     memblock_unref(chunk.memblock);
  82     sink_notify(c->sink_input->sink);
  83
  84     return 0;
  85 }
  86
  87 static int do_write(struct connection *c) {
  88     struct memchunk chunk;
  89     ssize_t r;
  90
  91     if (!iochannel_is_writable(c->io))
  92         return 0;
  93
  94     if (!c->source_output)
  95         return 0;
  96
  97     assert(c->output_memblockq);
  98     if (memblockq_peek(c->output_memblockq, &chunk) < 0)
  99         return 0;
 100
 101     assert(chunk.memblock && chunk.length);
 102
 103     if ((r = iochannel_write(c->io, chunk.memblock->data+chunk.index, chunk.length)) < 0) {
 104         fprintf(stderr, "write(): %s\n", strerror(errno));
 105         memblock_unref(chunk.memblock);
 106         return -1;
 107     }
 108
 109     memblockq_drop(c->output_memblockq, r);
 110     memblock_unref(chunk.memblock);
 111     return 0;
 112 }
 113
 114 /*** sink_input callbacks ***/
 115
 116 static int sink_input_peek_cb(struct sink_input *i, struct memchunk *chunk) {
 117     struct connection*c = i->userdata;
 118     assert(i && c && chunk);
 119
 120     if (memblockq_peek(c->input_memblockq, chunk) < 0)
 121         return -1;
 122
 123     return 0;
 124 }
 125
 126 static void sink_input_drop_cb(struct sink_input *i, size_t length) {
 127     struct connection*c = i->userdata;
 128     assert(i && c && length);
 129
 130     memblockq_drop(c->input_memblockq, length);
 131
 132     if (do_read(c) < 0)
 133         destroy_connection(c);
 134 }
 135
 136 static void sink_input_kill_cb(struct sink_input *i) {
 137     assert(i && i->userdata);
 138     destroy_connection((struct connection *) i->userdata);
 139 }
 140
 141
 142 static uint32_t sink_input_get_latency_cb(struct sink_input *i) {
 143     struct connection*c = i->userdata;
 144     assert(i && c);
 145     return samples_usec(memblockq_get_length(c->input_memblockq), &DEFAULT_SAMPLE_SPEC);
 146 }
 147
 148 /*** source_output callbacks ***/
 149
 150 static void source_output_push_cb(struct source_output *o, struct memchunk *chunk) {
 151     struct connection *c = o->userdata;
 152     assert(o && c && chunk);
 153
 154     memblockq_push(c->output_memblockq, chunk, 0);
 155
 156     if (do_write(c) < 0)
 157         destroy_connection(c);
 158 }
 159
 160 static void source_output_kill_cb(struct source_output *o) {
 161     assert(o && o->userdata);
 162     destroy_connection((struct connection *) o->userdata);
 163 }
 164
 165 /*** client callbacks ***/
 166
 167 static void client_kill_cb(struct client *c) {
 168     assert(c && c->userdata);
 169     destroy_connection((struct connection *) c->userdata);
 170 }
 171
 172 /*** iochannel callbacks ***/
 173
 174 static void io_callback(struct iochannel*io, void *userdata) {
 175     struct connection *c = userdata;
 176     assert(io && c && c->io == io);
 177
 178     if (do_read(c) < 0 || do_write(c) < 0)
 179         destroy_connection(c);
 180 }
 181
 182 /*** socket_server callbacks */
 183
 184 static void on_connection(struct socket_server*s, struct iochannel *io, void *userdata) {
 185     struct protocol_simple *p = userdata;
 186     struct connection *c = NULL;
 187     assert(s && io && p);
 188
 189     c = malloc(sizeof(struct connection));
 190     assert(c);
 191     c->io = io;
 192     c->sink_input = NULL;
 193     c->source_output = NULL;
 194     c->input_memblockq = c->output_memblockq = NULL;
 195     c->protocol = p;
 196
 197     c->client = client_new(p->core, "SIMPLE", "Client");
 198     assert(c->client);
 199     c->client->kill = client_kill_cb;
 200     c->client->userdata = c;
 201
 202     if (p->mode & PROTOCOL_SIMPLE_RECORD) {
 203         struct source *source;
 204         size_t l;
 205
 206         if (!(source = source_get_default(p->core))) {
 207             fprintf(stderr, "Failed to get default source.\n");
 208             goto fail;
 209         }
 210
 211         c->source_output = source_output_new(source, &DEFAULT_SAMPLE_SPEC, c->client->name);
 212         assert(c->source_output);
 213         c->source_output->push = source_output_push_cb;
 214         c->source_output->kill = source_output_kill_cb;
 215         c->source_output->userdata = c;
 216
 217         l = 5*bytes_per_second(&DEFAULT_SAMPLE_SPEC); /* 5s */
 218         c->output_memblockq = memblockq_new(l, sample_size(&DEFAULT_SAMPLE_SPEC), l/2);
 219     }
 220
 221     if (p->mode & PROTOCOL_SIMPLE_PLAYBACK) {
 222         struct sink *sink;
 223         size_t l;
 224
 225         if (!(sink = sink_get_default(p->core))) {
 226             fprintf(stderr, "Failed to get default sink.\n");
 227             goto fail;
 228         }
 229
 230         c->sink_input = sink_input_new(sink, &DEFAULT_SAMPLE_SPEC, c->client->name);
 231         assert(c->sink_input);
 232         c->sink_input->peek = sink_input_peek_cb;
 233         c->sink_input->drop = sink_input_drop_cb;
 234         c->sink_input->kill = sink_input_kill_cb;
 235         c->sink_input->get_latency = sink_input_get_latency_cb;
 236         c->sink_input->userdata = c;
 237
 238         l = bytes_per_second(&DEFAULT_SAMPLE_SPEC)/2; /* half a second */
 239         c->input_memblockq = memblockq_new(l, sample_size(&DEFAULT_SAMPLE_SPEC), l/2);
 240     }
 241
 242
 243     iochannel_set_callback(c->io, io_callback, c);
 244     idxset_put(p->connections, c, NULL);
 245     return;
 246
 247 fail:
 248     if (c) {
 249         free_connection(c, NULL);
 250         iochannel_free(c->io);
 251         free(c);
 252     }
 253 }
 254
 255 struct protocol_simple* protocol_simple_new(struct core *core, struct socket_server *server, enum protocol_simple_mode mode) {
 256     struct protocol_simple* p;
 257     assert(core && server && mode <= PROTOCOL_SIMPLE_DUPLEX && mode > 0);
 258
 259     p = malloc(sizeof(struct protocol_simple));
 260     assert(p);
 261     p->core = core;
 262     p->server = server;
 263     p->connections = idxset_new(NULL, NULL);
 264     p->mode = mode;
 265
 266     socket_server_set_callback(p->server, on_connection, p);
 267
 268     return p;
 269 }
 270
 271
 272 void protocol_simple_free(struct protocol_simple *p) {
 273     assert(p);
 274
 275     idxset_free(p->connections, free_connection, NULL);
 276     socket_server_free(p->server);
 277     free(p);
 278 }