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     size_t fragment_size;
  28 };
  29
  30 struct playback_stream {
  31     struct connection *connection;
  32     uint32_t index;
  33     struct pa_sink_input *sink_input;
  34     struct pa_memblockq *memblockq;
  35     size_t requested_bytes;
  36     int drain_request;
  37     uint32_t drain_tag;
  38 };
  39
  40 struct connection {
  41     int authorized;
  42     struct pa_protocol_native *protocol;
  43     struct pa_client *client;
  44     struct pa_pstream *pstream;
  45     struct pa_pdispatch *pdispatch;
  46     struct pa_idxset *record_streams, *playback_streams;
  47     uint32_t rrobin_index;
  48 };
  49
  50 struct pa_protocol_native {
  51     struct pa_module *module;
  52     int public;
  53     struct pa_core *core;
  54     struct pa_socket_server *server;
  55     struct pa_idxset *connections;
  56     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
  57 };
  58
  59 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk);
  60 static void sink_input_drop_cb(struct pa_sink_input *i, size_t length);
  61 static void sink_input_kill_cb(struct pa_sink_input *i);
  62 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i);
  63
  64 static void request_bytes(struct playback_stream*s);
  65
  66 static void source_output_kill_cb(struct pa_source_output *o);
  67 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk);
  68
  69 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  70 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  71 static void command_delete_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  72 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  73 static void command_create_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  74 static void command_delete_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  75 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  76 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  77 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  78
  79 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
  80     [PA_COMMAND_ERROR] = { NULL },
  81     [PA_COMMAND_TIMEOUT] = { NULL },
  82     [PA_COMMAND_REPLY] = { NULL },
  83     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { command_create_playback_stream },
  84     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { command_delete_playback_stream },
  85     [PA_COMMAND_DRAIN_PLAYBACK_STREAM] = { command_drain_playback_stream },
  86     [PA_COMMAND_CREATE_RECORD_STREAM] = { command_create_record_stream },
  87     [PA_COMMAND_DELETE_RECORD_STREAM] = { command_delete_record_stream },
  88     [PA_COMMAND_AUTH] = { command_auth },
  89     [PA_COMMAND_REQUEST] = { NULL },
  90     [PA_COMMAND_EXIT] = { command_exit },
  91     [PA_COMMAND_SET_NAME] = { command_set_name },
  92     [PA_COMMAND_LOOKUP_SINK] = { command_lookup },
  93     [PA_COMMAND_LOOKUP_SOURCE] = { command_lookup },
  94 };
  95
  96 /* structure management */
  97
  98 static struct record_stream* record_stream_new(struct connection *c, struct pa_source *source, struct pa_sample_spec *ss, const char *name, size_t maxlength, size_t fragment_size) {
  99     struct record_stream *s;
 100     struct pa_source_output *source_output;
 101     size_t base;
 102     assert(c && source && ss && name && maxlength);
 103
 104     if (!(source_output = pa_source_output_new(source, name, ss)))
 105         return NULL;
 106
 107     s = malloc(sizeof(struct record_stream));
 108     assert(s);
 109     s->connection = c;
 110     s->source_output = source_output;
 111     s->source_output->push = source_output_push_cb;
 112     s->source_output->kill = source_output_kill_cb;
 113     s->source_output->userdata = s;
 114     s->source_output->owner = c->protocol->module;
 115     s->source_output->client = c->client;
 116
 117     s->memblockq = pa_memblockq_new(maxlength, 0, base = pa_sample_size(ss), 0, 0);
 118     assert(s->memblockq);
 119
 120     s->fragment_size = (fragment_size/base)*base;
 121     if (!s->fragment_size)
 122         s->fragment_size = base;
 123
 124     pa_idxset_put(c->record_streams, s, &s->index);
 125     return s;
 126 }
 127
 128 static void record_stream_free(struct record_stream* r) {
 129     assert(r && r->connection);
 130
 131     pa_idxset_remove_by_data(r->connection->record_streams, r, NULL);
 132     pa_source_output_free(r->source_output);
 133     pa_memblockq_free(r->memblockq);
 134     free(r);
 135 }
 136
 137 static struct playback_stream* playback_stream_new(struct connection *c, struct pa_sink *sink, struct pa_sample_spec *ss, const char *name,
 138                                                    size_t maxlength,
 139                                                    size_t tlength,
 140                                                    size_t prebuf,
 141                                                    size_t minreq) {
 142     struct playback_stream *s;
 143     struct pa_sink_input *sink_input;
 144     assert(c && sink && ss && name && maxlength);
 145
 146     if (!(sink_input = pa_sink_input_new(sink, name, ss)))
 147         return NULL;
 148
 149     s = malloc(sizeof(struct playback_stream));
 150     assert (s);
 151     s->connection = c;
 152     s->sink_input = sink_input;
 153
 154     s->sink_input->peek = sink_input_peek_cb;
 155     s->sink_input->drop = sink_input_drop_cb;
 156     s->sink_input->kill = sink_input_kill_cb;
 157     s->sink_input->get_latency = sink_input_get_latency_cb;
 158     s->sink_input->userdata = s;
 159     s->sink_input->owner = c->protocol->module;
 160     s->sink_input->client = c->client;
 161
 162     s->memblockq = pa_memblockq_new(maxlength, tlength, pa_sample_size(ss), prebuf, minreq);
 163     assert(s->memblockq);
 164
 165     s->requested_bytes = 0;
 166     s->drain_request = 0;
 167
 168     pa_idxset_put(c->playback_streams, s, &s->index);
 169     return s;
 170 }
 171
 172 static void playback_stream_free(struct playback_stream* p) {
 173     assert(p && p->connection);
 174
 175     if (p->drain_request)
 176         pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERROR_NOENTITY);
 177
 178     pa_idxset_remove_by_data(p->connection->playback_streams, p, NULL);
 179     pa_sink_input_free(p->sink_input);
 180     pa_memblockq_free(p->memblockq);
 181     free(p);
 182 }
 183
 184 static void connection_free(struct connection *c) {
 185     struct record_stream *r;
 186     struct playback_stream *p;
 187     assert(c && c->protocol);
 188
 189     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 190     while ((r = pa_idxset_first(c->record_streams, NULL)))
 191         record_stream_free(r);
 192     pa_idxset_free(c->record_streams, NULL, NULL);
 193
 194     while ((p = pa_idxset_first(c->playback_streams, NULL)))
 195         playback_stream_free(p);
 196     pa_idxset_free(c->playback_streams, NULL, NULL);
 197
 198     pa_pdispatch_free(c->pdispatch);
 199     pa_pstream_free(c->pstream);
 200     pa_client_free(c->client);
 201     free(c);
 202 }
 203
 204 static void request_bytes(struct playback_stream *s) {
 205     struct pa_tagstruct *t;
 206     size_t l;
 207     assert(s);
 208
 209     if (!(l = pa_memblockq_missing(s->memblockq)))
 210         return;
 211
 212     if (l <= s->requested_bytes)
 213         return;
 214
 215     l -= s->requested_bytes;
 216
 217     if (l < pa_memblockq_get_minreq(s->memblockq))
 218         return;
 219
 220     s->requested_bytes += l;
 221
 222     t = pa_tagstruct_new(NULL, 0);
 223     assert(t);
 224     pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 225     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 226     pa_tagstruct_putu32(t, s->index);
 227     pa_tagstruct_putu32(t, l);
 228     pa_pstream_send_tagstruct(s->connection->pstream, t);
 229
 230     /*fprintf(stderr, "Requesting %u bytes\n", l);*/
 231 }
 232
 233 static void send_memblock(struct connection *c) {
 234     uint32_t start;
 235     struct record_stream *r;
 236
 237     start = PA_IDXSET_INVALID;
 238     for (;;) {
 239         struct pa_memchunk chunk;
 240
 241         if (!(r = pa_idxset_rrobin(c->record_streams, &c->rrobin_index)))
 242             return;
 243
 244         if (start == PA_IDXSET_INVALID)
 245             start = c->rrobin_index;
 246         else if (start == c->rrobin_index)
 247             return;
 248
 249         if (pa_memblockq_peek(r->memblockq,  &chunk) >= 0) {
 250             if (chunk.length > r->fragment_size)
 251                 chunk.length = r->fragment_size;
 252
 253             pa_pstream_send_memblock(c->pstream, r->index, 0, &chunk);
 254             pa_memblockq_drop(r->memblockq, chunk.length);
 255             pa_memblock_unref(chunk.memblock);
 256
 257             return;
 258         }
 259     }
 260 }
 261
 262 static void send_playback_stream_killed(struct playback_stream *p) {
 263     struct pa_tagstruct *t;
 264     assert(p);
 265
 266     t = pa_tagstruct_new(NULL, 0);
 267     assert(t);
 268     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_KILLED);
 269     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 270     pa_tagstruct_putu32(t, p->index);
 271     pa_pstream_send_tagstruct(p->connection->pstream, t);
 272 }
 273
 274 static void send_record_stream_killed(struct record_stream *r) {
 275     struct pa_tagstruct *t;
 276     assert(r);
 277
 278     t = pa_tagstruct_new(NULL, 0);
 279     assert(t);
 280     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_KILLED);
 281     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 282     pa_tagstruct_putu32(t, r->index);
 283     pa_pstream_send_tagstruct(r->connection->pstream, t);
 284 }
 285
 286
 287 /*** sinkinput callbacks ***/
 288
 289 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk) {
 290     struct playback_stream *s;
 291     assert(i && i->userdata && chunk);
 292     s = i->userdata;
 293
 294     if (pa_memblockq_peek(s->memblockq, chunk) < 0)
 295         return -1;
 296
 297     return 0;
 298 }
 299
 300 static void sink_input_drop_cb(struct pa_sink_input *i, size_t length) {
 301     struct playback_stream *s;
 302     assert(i && i->userdata && length);
 303     s = i->userdata;
 304
 305     pa_memblockq_drop(s->memblockq, length);
 306     request_bytes(s);
 307
 308     if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) {
 309         pa_pstream_send_simple_ack(s->connection->pstream, s->drain_tag);
 310         s->drain_request = 0;
 311     }
 312 }
 313
 314 static void sink_input_kill_cb(struct pa_sink_input *i) {
 315     assert(i && i->userdata);
 316     send_playback_stream_killed((struct playback_stream *) i->userdata);
 317     playback_stream_free((struct playback_stream *) i->userdata);
 318 }
 319
 320 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i) {
 321     struct playback_stream *s;
 322     assert(i && i->userdata);
 323     s = i->userdata;
 324
 325     return pa_samples_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
 326 }
 327
 328 /*** source_output callbacks ***/
 329
 330 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk) {
 331     struct record_stream *s;
 332     assert(o && o->userdata && chunk);
 333     s = o->userdata;
 334
 335     pa_memblockq_push(s->memblockq, chunk, 0);
 336     if (!pa_pstream_is_pending(s->connection->pstream))
 337         send_memblock(s->connection);
 338 }
 339
 340 static void source_output_kill_cb(struct pa_source_output *o) {
 341     assert(o && o->userdata);
 342     send_record_stream_killed((struct record_stream *) o->userdata);
 343     record_stream_free((struct record_stream *) o->userdata);
 344 }
 345
 346 /*** pdispatch callbacks ***/
 347
 348 static void protocol_error(struct connection *c) {
 349     fprintf(stderr, __FILE__": protocol error, kicking client\n");
 350     connection_free(c);
 351 }
 352
 353 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 354     struct connection *c = userdata;
 355     struct playback_stream *s;
 356     size_t maxlength, tlength, prebuf, minreq;
 357     uint32_t sink_index;
 358     const char *name;
 359     struct pa_sample_spec ss;
 360     struct pa_tagstruct *reply;
 361     struct pa_sink *sink;
 362     assert(c && t && c->protocol && c->protocol->core);
 363
 364     if (pa_tagstruct_gets(t, &name) < 0 ||
 365         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 366         pa_tagstruct_getu32(t, &sink_index) < 0 ||
 367         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 368         pa_tagstruct_getu32(t, &tlength) < 0 ||
 369         pa_tagstruct_getu32(t, &prebuf) < 0 ||
 370         pa_tagstruct_getu32(t, &minreq) < 0 ||
 371         !pa_tagstruct_eof(t)) {
 372         protocol_error(c);
 373         return;
 374     }
 375
 376     if (!c->authorized) {
 377         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 378         return;
 379     }
 380
 381     if (sink_index == (uint32_t) -1)
 382         sink = pa_sink_get_default(c->protocol->core);
 383     else
 384         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
 385
 386     if (!sink) {
 387         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 388         return;
 389     }
 390
 391     if (!(s = playback_stream_new(c, sink, &ss, name, maxlength, tlength, prebuf, minreq))) {
 392         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 393         return;
 394     }
 395
 396     reply = pa_tagstruct_new(NULL, 0);
 397     assert(reply);
 398     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 399     pa_tagstruct_putu32(reply, tag);
 400     pa_tagstruct_putu32(reply, s->index);
 401     assert(s->sink_input);
 402     pa_tagstruct_putu32(reply, s->sink_input->index);
 403     pa_pstream_send_tagstruct(c->pstream, reply);
 404     request_bytes(s);
 405 }
 406
 407 static void command_delete_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 408     struct connection *c = userdata;
 409     uint32_t channel;
 410     struct playback_stream *s;
 411     assert(c && t);
 412
 413     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 414         !pa_tagstruct_eof(t)) {
 415         protocol_error(c);
 416         return;
 417     }
 418
 419     if (!c->authorized) {
 420         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 421         return;
 422     }
 423
 424     if (!(s = pa_idxset_get_by_index(c->playback_streams, channel))) {
 425         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 426         return;
 427     }
 428
 429     playback_stream_free(s);
 430     pa_pstream_send_simple_ack(c->pstream, tag);
 431 }
 432
 433 static void command_create_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 434     struct connection *c = userdata;
 435     struct record_stream *s;
 436     size_t maxlength, fragment_size;
 437     uint32_t source_index;
 438     const char *name;
 439     struct pa_sample_spec ss;
 440     struct pa_tagstruct *reply;
 441     struct pa_source *source;
 442     assert(c && t && c->protocol && c->protocol->core);
 443
 444     if (pa_tagstruct_gets(t, &name) < 0 ||
 445         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 446         pa_tagstruct_getu32(t, &source_index) < 0 ||
 447         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 448         pa_tagstruct_getu32(t, &fragment_size) < 0 ||
 449         !pa_tagstruct_eof(t)) {
 450         protocol_error(c);
 451         return;
 452     }
 453
 454     if (!c->authorized) {
 455         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 456         return;
 457     }
 458
 459     if (source_index == (uint32_t) -1)
 460         source = pa_source_get_default(c->protocol->core);
 461     else
 462         source = pa_idxset_get_by_index(c->protocol->core->sources, source_index);
 463
 464     if (!source) {
 465         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 466         return;
 467     }
 468
 469     if (!(s = record_stream_new(c, source, &ss, name, maxlength, fragment_size))) {
 470         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 471         return;
 472     }
 473
 474     reply = pa_tagstruct_new(NULL, 0);
 475     assert(reply);
 476     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 477     pa_tagstruct_putu32(reply, tag);
 478     pa_tagstruct_putu32(reply, s->index);
 479     assert(s->source_output);
 480     pa_tagstruct_putu32(reply, s->source_output->index);
 481     pa_pstream_send_tagstruct(c->pstream, reply);
 482 }
 483
 484 static void command_delete_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 485     struct connection *c = userdata;
 486     uint32_t channel;
 487     struct record_stream *s;
 488     assert(c && t);
 489
 490     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 491         !pa_tagstruct_eof(t)) {
 492         protocol_error(c);
 493         return;
 494     }
 495
 496     if (!c->authorized) {
 497         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 498         return;
 499     }
 500
 501     if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
 502         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 503         return;
 504     }
 505
 506     record_stream_free(s);
 507     pa_pstream_send_simple_ack(c->pstream, tag);
 508 }
 509
 510 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 511     struct connection *c = userdata;
 512     assert(c && t);
 513
 514     if (!pa_tagstruct_eof(t)) {
 515         protocol_error(c);
 516         return;
 517     }
 518
 519     if (!c->authorized) {
 520         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 521         return;
 522     }
 523
 524     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop);
 525     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
 526     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
 527     return;
 528 }
 529
 530 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 531     struct connection *c = userdata;
 532     const void*cookie;
 533     assert(c && t);
 534
 535     if (pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
 536         !pa_tagstruct_eof(t)) {
 537         protocol_error(c);
 538         return;
 539     }
 540
 541     if (memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) != 0) {
 542         fprintf(stderr, "protocol-native.c: Denied access to client with invalid authorization key.\n");
 543         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 544         return;
 545     }
 546
 547     c->authorized = 1;
 548     pa_pstream_send_simple_ack(c->pstream, tag);
 549     return;
 550 }
 551
 552 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 553     struct connection *c = userdata;
 554     const char *name;
 555     assert(c && t);
 556
 557     if (pa_tagstruct_gets(t, &name) < 0 ||
 558         !pa_tagstruct_eof(t)) {
 559         protocol_error(c);
 560         return;
 561     }
 562
 563     pa_client_rename(c->client, name);
 564     pa_pstream_send_simple_ack(c->pstream, tag);
 565     return;
 566 }
 567
 568 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 569     struct connection *c = userdata;
 570     const char *name;
 571     uint32_t index = PA_IDXSET_INVALID;
 572     assert(c && t);
 573
 574     if (pa_tagstruct_gets(t, &name) < 0 ||
 575         !pa_tagstruct_eof(t)) {
 576         protocol_error(c);
 577         return;
 578     }
 579
 580     if (!c->authorized) {
 581         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 582         return;
 583     }
 584
 585     if (command == PA_COMMAND_LOOKUP_SINK) {
 586         struct pa_sink *sink;
 587         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK)))
 588             index = sink->index;
 589     } else {
 590         struct pa_source *source;
 591         assert(command == PA_COMMAND_LOOKUP_SOURCE);
 592         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE)))
 593             index = source->index;
 594     }
 595
 596     if (index == PA_IDXSET_INVALID)
 597         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 598     else {
 599         struct pa_tagstruct *reply;
 600         reply = pa_tagstruct_new(NULL, 0);
 601         assert(reply);
 602         pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 603         pa_tagstruct_putu32(reply, tag);
 604         pa_tagstruct_putu32(reply, index);
 605         pa_pstream_send_tagstruct(c->pstream, reply);
 606     }
 607 }
 608
 609 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 610     struct connection *c = userdata;
 611     uint32_t index;
 612     struct playback_stream *s;
 613     assert(c && t);
 614
 615     if (pa_tagstruct_getu32(t, &index) < 0 ||
 616         !pa_tagstruct_eof(t)) {
 617         protocol_error(c);
 618         return;
 619     }
 620
 621     if (!c->authorized) {
 622         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 623         return;
 624     }
 625
 626     if (!(s = pa_idxset_get_by_index(c->playback_streams, index))) {
 627         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 628         return;
 629     }
 630
 631     s->drain_request = 0;
 632
 633     if (!pa_memblockq_is_readable(s->memblockq))
 634         pa_pstream_send_simple_ack(c->pstream, tag);
 635     else {
 636         s->drain_request = 1;
 637         s->drain_tag = tag;
 638     }
 639 }
 640
 641 /*** pstream callbacks ***/
 642
 643 static void pstream_packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
 644     struct connection *c = userdata;
 645     assert(p && packet && packet->data && c);
 646
 647     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
 648         fprintf(stderr, "protocol-native: invalid packet.\n");
 649         connection_free(c);
 650     }
 651 }
 652
 653 static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, int32_t delta, const struct pa_memchunk *chunk, void *userdata) {
 654     struct connection *c = userdata;
 655     struct playback_stream *stream;
 656     assert(p && chunk && userdata);
 657
 658     if (!(stream = pa_idxset_get_by_index(c->playback_streams, channel))) {
 659         fprintf(stderr, "protocol-native: client sent block for invalid stream.\n");
 660         connection_free(c);
 661         return;
 662     }
 663
 664     if (chunk->length >= stream->requested_bytes)
 665         stream->requested_bytes = 0;
 666     else
 667         stream->requested_bytes -= chunk->length;
 668
 669     pa_memblockq_push_align(stream->memblockq, chunk, delta);
 670     assert(stream->sink_input);
 671     pa_sink_notify(stream->sink_input->sink);
 672
 673     /*fprintf(stderr, "Recieved %u bytes.\n", chunk->length);*/
 674 }
 675
 676 static void pstream_die_callback(struct pa_pstream *p, void *userdata) {
 677     struct connection *c = userdata;
 678     assert(p && c);
 679     connection_free(c);
 680
 681     fprintf(stderr, "protocol-native: connection died.\n");
 682 }
 683
 684
 685 static void pstream_drain_callback(struct pa_pstream *p, void *userdata) {
 686     struct connection *c = userdata;
 687     assert(p && c);
 688
 689     send_memblock(c);
 690 }
 691
 692 /*** client callbacks ***/
 693
 694 static void client_kill_cb(struct pa_client *c) {
 695     assert(c && c->userdata);
 696     connection_free(c->userdata);
 697 }
 698
 699 /*** socket server callbacks ***/
 700
 701 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
 702     struct pa_protocol_native *p = userdata;
 703     struct connection *c;
 704     assert(s && io && p);
 705
 706     c = malloc(sizeof(struct connection));
 707     assert(c);
 708     c->authorized = p->public;
 709     c->protocol = p;
 710     assert(p->core);
 711     c->client = pa_client_new(p->core, "NATIVE", "Client");
 712     assert(c->client);
 713     c->client->kill = client_kill_cb;
 714     c->client->userdata = c;
 715     c->client->owner = p->module;
 716
 717     c->pstream = pa_pstream_new(p->core->mainloop, io);
 718     assert(c->pstream);
 719
 720     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
 721     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
 722     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
 723     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
 724
 725     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
 726     assert(c->pdispatch);
 727
 728     c->record_streams = pa_idxset_new(NULL, NULL);
 729     c->playback_streams = pa_idxset_new(NULL, NULL);
 730     assert(c->record_streams && c->playback_streams);
 731
 732     c->rrobin_index = PA_IDXSET_INVALID;
 733
 734     pa_idxset_put(p->connections, c, NULL);
 735 }
 736
 737 /*** module entry points ***/
 738
 739 struct pa_protocol_native* pa_protocol_native_new(struct pa_core *core, struct pa_socket_server *server, struct pa_module *m) {
 740     struct pa_protocol_native *p;
 741     assert(core && server);
 742
 743     p = malloc(sizeof(struct pa_protocol_native));
 744     assert(p);
 745
 746     if (pa_authkey_load_from_home(PA_NATIVE_COOKIE_FILE, p->auth_cookie, sizeof(p->auth_cookie)) < 0) {
 747         free(p);
 748         return NULL;
 749     }
 750
 751     p->module = m;
 752     p->public = 1;
 753     p->server = server;
 754     p->core = core;
 755     p->connections = pa_idxset_new(NULL, NULL);
 756     assert(p->connections);
 757
 758     pa_socket_server_set_callback(p->server, on_connection, p);
 759
 760     return p;
 761 }
 762
 763 void pa_protocol_native_free(struct pa_protocol_native *p) {
 764     struct connection *c;
 765     assert(p);
 766
 767     while ((c = pa_idxset_first(p->connections, NULL)))
 768         connection_free(c);
 769     pa_idxset_free(p->connections, NULL, NULL);
 770     pa_socket_server_free(p->server);
 771     free(p);
 772 }