src/protocol-native.c

   1 #include <string.h>
   2 #include <stdio.h>
   3 #include <assert.h>
   4 #include <stdlib.h>
   5
   6 #include "protocol-native.h"
   7 #include "protocol-native-spec.h"
   8 #include "packet.h"
   9 #include "client.h"
  10 #include "sourceoutput.h"
  11 #include "sinkinput.h"
  12 #include "pstream.h"
  13 #include "tagstruct.h"
  14 #include "pdispatch.h"
  15 #include "pstream-util.h"
  16 #include "authkey.h"
  17 #include "namereg.h"
  18
  19 struct connection;
  20 struct pa_protocol_native;
  21
  22 struct record_stream {
  23     struct connection *connection;
  24     uint32_t index;
  25     struct pa_source_output *source_output;
  26     struct pa_memblockq *memblockq;
  27 };
  28
  29 struct playback_stream {
  30     struct connection *connection;
  31     uint32_t index;
  32     struct pa_sink_input *sink_input;
  33     struct pa_memblockq *memblockq;
  34     size_t requested_bytes;
  35 };
  36
  37 struct connection {
  38     int authorized;
  39     struct pa_protocol_native *protocol;
  40     struct pa_client *client;
  41     struct pa_pstream *pstream;
  42     struct pa_pdispatch *pdispatch;
  43     struct pa_idxset *record_streams, *playback_streams;
  44 };
  45
  46 struct pa_protocol_native {
  47     int public;
  48     struct pa_core *core;
  49     struct pa_socket_server *server;
  50     struct pa_idxset *connections;
  51     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
  52 };
  53
  54 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk);
  55 static void sink_input_drop_cb(struct pa_sink_input *i, size_t length);
  56 static void sink_input_kill_cb(struct pa_sink_input *i);
  57 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i);
  58
  59 static void request_bytes(struct playback_stream*s);
  60
  61 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  62 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  63 static void command_delete_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  64 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  65 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  66 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  67
  68 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
  69     [PA_COMMAND_ERROR] = { NULL },
  70     [PA_COMMAND_TIMEOUT] = { NULL },
  71     [PA_COMMAND_REPLY] = { NULL },
  72     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { command_create_playback_stream },
  73     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { command_delete_playback_stream },
  74     [PA_COMMAND_CREATE_RECORD_STREAM] = { NULL },
  75     [PA_COMMAND_DELETE_RECORD_STREAM] = { NULL },
  76     [PA_COMMAND_AUTH] = { command_auth },
  77     [PA_COMMAND_REQUEST] = { NULL },
  78     [PA_COMMAND_EXIT] = { command_exit },
  79     [PA_COMMAND_SET_NAME] = { command_set_name },
  80     [PA_COMMAND_LOOKUP_SINK] = { command_lookup },
  81     [PA_COMMAND_LOOKUP_SOURCE] = { command_lookup },
  82 };
  83
  84 /* structure management */
  85
  86 static void record_stream_free(struct record_stream* r) {
  87     assert(r && r->connection);
  88
  89     pa_idxset_remove_by_data(r->connection->record_streams, r, NULL);
  90     pa_source_output_free(r->source_output);
  91     pa_memblockq_free(r->memblockq);
  92     free(r);
  93 }
  94
  95 static struct playback_stream* playback_stream_new(struct connection *c, struct pa_sink *sink, struct pa_sample_spec *ss, const char *name,
  96                                                    size_t maxlength,
  97                                                    size_t tlength,
  98                                                    size_t prebuf,
  99                                                    size_t minreq) {
 100     struct playback_stream *s;
 101     assert(c && sink && ss && name && maxlength);
 102
 103     s = malloc(sizeof(struct playback_stream));
 104     assert (s);
 105     s->connection = c;
 106
 107     s->sink_input = pa_sink_input_new(sink, name, ss);
 108     assert(s->sink_input);
 109     s->sink_input->peek = sink_input_peek_cb;
 110     s->sink_input->drop = sink_input_drop_cb;
 111     s->sink_input->kill = sink_input_kill_cb;
 112     s->sink_input->get_latency = sink_input_get_latency_cb;
 113     s->sink_input->userdata = s;
 114
 115     s->memblockq = pa_memblockq_new(maxlength, tlength, pa_sample_size(ss), prebuf, minreq);
 116     assert(s->memblockq);
 117
 118     s->requested_bytes = 0;
 119
 120     pa_idxset_put(c->playback_streams, s, &s->index);
 121     return s;
 122 }
 123
 124 static void playback_stream_free(struct playback_stream* p) {
 125     assert(p && p->connection);
 126
 127     pa_idxset_remove_by_data(p->connection->playback_streams, p, NULL);
 128     pa_sink_input_free(p->sink_input);
 129     pa_memblockq_free(p->memblockq);
 130     free(p);
 131 }
 132
 133 static void connection_free(struct connection *c) {
 134     struct record_stream *r;
 135     struct playback_stream *p;
 136     assert(c && c->protocol);
 137
 138     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 139     while ((r = pa_idxset_first(c->record_streams, NULL)))
 140         record_stream_free(r);
 141     pa_idxset_free(c->record_streams, NULL, NULL);
 142
 143     while ((p = pa_idxset_first(c->playback_streams, NULL)))
 144         playback_stream_free(p);
 145     pa_idxset_free(c->playback_streams, NULL, NULL);
 146
 147     pa_pdispatch_free(c->pdispatch);
 148     pa_pstream_free(c->pstream);
 149     pa_client_free(c->client);
 150     free(c);
 151 }
 152
 153 static void request_bytes(struct playback_stream *s) {
 154     struct pa_tagstruct *t;
 155     size_t l;
 156     assert(s);
 157
 158     if (!(l = pa_memblockq_missing(s->memblockq)))
 159         return;
 160
 161     if (l <= s->requested_bytes)
 162         return;
 163
 164     l -= s->requested_bytes;
 165
 166     if (l < pa_memblockq_get_minreq(s->memblockq))
 167         return;
 168
 169     s->requested_bytes += l;
 170
 171     t = pa_tagstruct_new(NULL, 0);
 172     assert(t);
 173     pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 174     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 175     pa_tagstruct_putu32(t, s->index);
 176     pa_tagstruct_putu32(t, l);
 177     pa_pstream_send_tagstruct(s->connection->pstream, t);
 178
 179     /*fprintf(stderr, "Requesting %u bytes\n", l);*/
 180 }
 181
 182 /*** sinkinput callbacks ***/
 183
 184 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk) {
 185     struct playback_stream *s;
 186     assert(i && i->userdata && chunk);
 187     s = i->userdata;
 188
 189     if (pa_memblockq_peek(s->memblockq, chunk) < 0)
 190         return -1;
 191
 192     return 0;
 193 }
 194
 195 static void sink_input_drop_cb(struct pa_sink_input *i, size_t length) {
 196     struct playback_stream *s;
 197     assert(i && i->userdata && length);
 198     s = i->userdata;
 199
 200     pa_memblockq_drop(s->memblockq, length);
 201     request_bytes(s);
 202 }
 203
 204 static void sink_input_kill_cb(struct pa_sink_input *i) {
 205     struct playback_stream *s;
 206     assert(i && i->userdata);
 207     s = i->userdata;
 208
 209     playback_stream_free(s);
 210 }
 211
 212 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i) {
 213     struct playback_stream *s;
 214     assert(i && i->userdata);
 215     s = i->userdata;
 216
 217     return pa_samples_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
 218 }
 219
 220 /*** pdispatch callbacks ***/
 221
 222 static void protocol_error(struct connection *c) {
 223     fprintf(stderr, __FILE__": protocol error, kicking client\n");
 224     connection_free(c);
 225 }
 226
 227 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 228     struct connection *c = userdata;
 229     struct playback_stream *s;
 230     size_t maxlength, tlength, prebuf, minreq;
 231     uint32_t sink_index;
 232     const char *name;
 233     struct pa_sample_spec ss;
 234     struct pa_tagstruct *reply;
 235     struct pa_sink *sink;
 236     assert(c && t && c->protocol && c->protocol->core);
 237
 238     if (pa_tagstruct_gets(t, &name) < 0 ||
 239         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 240         pa_tagstruct_getu32(t, &sink_index) < 0 ||
 241         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 242         pa_tagstruct_getu32(t, &tlength) < 0 ||
 243         pa_tagstruct_getu32(t, &prebuf) < 0 ||
 244         pa_tagstruct_getu32(t, &minreq) < 0 ||
 245         !pa_tagstruct_eof(t)) {
 246         protocol_error(c);
 247         return;
 248     }
 249
 250     if (!c->authorized) {
 251         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 252         return;
 253     }
 254
 255     if (sink_index == (uint32_t) -1)
 256         sink = pa_sink_get_default(c->protocol->core);
 257     else
 258         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
 259
 260     if (!sink) {
 261         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 262         return;
 263     }
 264
 265     if (!(s = playback_stream_new(c, sink, &ss, name, maxlength, tlength, prebuf, minreq))) {
 266         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 267         return;
 268     }
 269
 270     reply = pa_tagstruct_new(NULL, 0);
 271     assert(reply);
 272     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 273     pa_tagstruct_putu32(reply, tag);
 274     pa_tagstruct_putu32(reply, s->index);
 275     assert(s->sink_input);
 276     pa_tagstruct_putu32(reply, s->sink_input->index);
 277     pa_pstream_send_tagstruct(c->pstream, reply);
 278     request_bytes(s);
 279 }
 280
 281 static void command_delete_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 282     struct connection *c = userdata;
 283     uint32_t channel;
 284     struct playback_stream *s;
 285     assert(c && t);
 286
 287     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 288         !pa_tagstruct_eof(t)) {
 289         protocol_error(c);
 290         return;
 291     }
 292
 293     if (!c->authorized) {
 294         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 295         return;
 296     }
 297
 298     if (!(s = pa_idxset_get_by_index(c->playback_streams, channel))) {
 299         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 300         return;
 301     }
 302
 303     pa_pstream_send_simple_ack(c->pstream, tag);
 304 }
 305
 306 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 307     struct connection *c = userdata;
 308     assert(c && t);
 309
 310     if (!pa_tagstruct_eof(t)) {
 311         protocol_error(c);
 312         return;
 313     }
 314
 315     if (!c->authorized) {
 316         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 317         return;
 318     }
 319
 320     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop);
 321     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
 322     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
 323     return;
 324 }
 325
 326 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 327     struct connection *c = userdata;
 328     const void*cookie;
 329     assert(c && t);
 330
 331     if (pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
 332         !pa_tagstruct_eof(t)) {
 333         protocol_error(c);
 334         return;
 335     }
 336
 337     if (memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) != 0) {
 338         fprintf(stderr, "protocol-native.c: Denied access to client with invalid authorization key.\n");
 339         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 340         return;
 341     }
 342
 343     c->authorized = 1;
 344     pa_pstream_send_simple_ack(c->pstream, tag);
 345     return;
 346 }
 347
 348 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 349     struct connection *c = userdata;
 350     const char *name;
 351     assert(c && t);
 352
 353     if (pa_tagstruct_gets(t, &name) < 0 ||
 354         !pa_tagstruct_eof(t)) {
 355         protocol_error(c);
 356         return;
 357     }
 358
 359     pa_client_rename(c->client, name);
 360     pa_pstream_send_simple_ack(c->pstream, tag);
 361     return;
 362 }
 363
 364 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 365     struct connection *c = userdata;
 366     const char *name;
 367     uint32_t index = PA_IDXSET_INVALID;
 368     assert(c && t);
 369
 370     if (pa_tagstruct_gets(t, &name) < 0 ||
 371         !pa_tagstruct_eof(t)) {
 372         protocol_error(c);
 373         return;
 374     }
 375
 376     if (command == PA_COMMAND_LOOKUP_SINK) {
 377         struct pa_sink *sink;
 378         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK)))
 379             index = sink->index;
 380     } else {
 381         struct pa_source *source;
 382         assert(command == PA_COMMAND_LOOKUP_SOURCE);
 383         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE)))
 384             index = source->index;
 385     }
 386
 387     if (index == PA_IDXSET_INVALID)
 388         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 389     else {
 390         struct pa_tagstruct *reply;
 391         reply = pa_tagstruct_new(NULL, 0);
 392         assert(reply);
 393         pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 394         pa_tagstruct_putu32(reply, tag);
 395         pa_tagstruct_putu32(reply, index);
 396         pa_pstream_send_tagstruct(c->pstream, reply);
 397     }
 398 }
 399
 400 /*** pstream callbacks ***/
 401
 402 static void packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
 403     struct connection *c = userdata;
 404     assert(p && packet && packet->data && c);
 405
 406     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
 407         fprintf(stderr, "protocol-native: invalid packet.\n");
 408         connection_free(c);
 409     }
 410 }
 411
 412 static void memblock_callback(struct pa_pstream *p, uint32_t channel, int32_t delta, struct pa_memchunk *chunk, void *userdata) {
 413     struct connection *c = userdata;
 414     struct playback_stream *stream;
 415     assert(p && chunk && userdata);
 416
 417     if (!(stream = pa_idxset_get_by_index(c->playback_streams, channel))) {
 418         fprintf(stderr, "protocol-native: client sent block for invalid stream.\n");
 419         connection_free(c);
 420         return;
 421     }
 422
 423     if (chunk->length >= stream->requested_bytes)
 424         stream->requested_bytes = 0;
 425     else
 426         stream->requested_bytes -= chunk->length;
 427
 428     pa_memblockq_push_align(stream->memblockq, chunk, delta);
 429     assert(stream->sink_input);
 430     pa_sink_notify(stream->sink_input->sink);
 431
 432     /*fprintf(stderr, "Recieved %u bytes.\n", chunk->length);*/
 433 }
 434
 435 static void die_callback(struct pa_pstream *p, void *userdata) {
 436     struct connection *c = userdata;
 437     assert(p && c);
 438     connection_free(c);
 439
 440     fprintf(stderr, "protocol-native: connection died.\n");
 441 }
 442
 443 /*** client callbacks ***/
 444
 445 static void client_kill_cb(struct pa_client *c) {
 446     assert(c && c->userdata);
 447     connection_free(c->userdata);
 448 }
 449
 450 /*** socket server callbacks ***/
 451
 452 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
 453     struct pa_protocol_native *p = userdata;
 454     struct connection *c;
 455     assert(s && io && p);
 456
 457     c = malloc(sizeof(struct connection));
 458     assert(c);
 459     c->authorized = p->public;
 460     c->protocol = p;
 461     assert(p->core);
 462     c->client = pa_client_new(p->core, "NATIVE", "Client");
 463     assert(c->client);
 464     c->client->kill = client_kill_cb;
 465     c->client->userdata = c;
 466     c->pstream = pa_pstream_new(p->core->mainloop, io);
 467     assert(c->pstream);
 468
 469     pa_pstream_set_recieve_packet_callback(c->pstream, packet_callback, c);
 470     pa_pstream_set_recieve_memblock_callback(c->pstream, memblock_callback, c);
 471     pa_pstream_set_die_callback(c->pstream, die_callback, c);
 472
 473     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
 474     assert(c->pdispatch);
 475
 476     c->record_streams = pa_idxset_new(NULL, NULL);
 477     c->playback_streams = pa_idxset_new(NULL, NULL);
 478     assert(c->record_streams && c->playback_streams);
 479
 480     pa_idxset_put(p->connections, c, NULL);
 481 }
 482
 483 /*** module entry points ***/
 484
 485 struct pa_protocol_native* pa_protocol_native_new(struct pa_core *core, struct pa_socket_server *server) {
 486     struct pa_protocol_native *p;
 487     assert(core && server);
 488
 489     p = malloc(sizeof(struct pa_protocol_native));
 490     assert(p);
 491
 492     if (pa_authkey_load_from_home(PA_NATIVE_COOKIE_FILE, p->auth_cookie, sizeof(p->auth_cookie)) < 0) {
 493         free(p);
 494         return NULL;
 495     }
 496
 497     p->public = 1;
 498     p->server = server;
 499     p->core = core;
 500     p->connections = pa_idxset_new(NULL, NULL);
 501     assert(p->connections);
 502
 503     pa_socket_server_set_callback(p->server, on_connection, p);
 504
 505     return p;
 506 }
 507
 508 void pa_protocol_native_free(struct pa_protocol_native *p) {
 509     struct connection *c;
 510     assert(p);
 511
 512     while ((c = pa_idxset_first(p->connections, NULL)))
 513         connection_free(c);
 514     pa_idxset_free(p->connections, NULL, NULL);
 515     pa_socket_server_free(p->server);
 516     free(p);
 517 }