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