polyp/protocol-native.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of polypaudio.
   5
   6   polypaudio is free software; you can redistribute it and/or modify
   7   it under the terms of the GNU General Public License as published
   8   by the Free Software Foundation; either version 2 of the License,
   9   or (at your option) any later version.
  10
  11   polypaudio is distributed in the hope that it will be useful, but
  12   WITHOUT ANY WARRANTY; without even the implied warranty of
  13   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14   General Public License for more details.
  15
  16   You should have received a copy of the GNU General Public License
  17   along with polypaudio; if not, write to the Free Software
  18   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  19   USA.
  20 ***/
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include <string.h>
  27 #include <stdio.h>
  28 #include <assert.h>
  29 #include <stdlib.h>
  30
  31 #include "protocol-native.h"
  32 #include "native-common.h"
  33 #include "packet.h"
  34 #include "client.h"
  35 #include "source-output.h"
  36 #include "sink-input.h"
  37 #include "pstream.h"
  38 #include "tagstruct.h"
  39 #include "pdispatch.h"
  40 #include "pstream-util.h"
  41 #include "authkey.h"
  42 #include "namereg.h"
  43 #include "scache.h"
  44 #include "xmalloc.h"
  45 #include "util.h"
  46 #include "subscribe.h"
  47 #include "log.h"
  48
  49 struct connection;
  50 struct pa_protocol_native;
  51
  52 struct record_stream {
  53     struct connection *connection;
  54     uint32_t index;
  55     struct pa_source_output *source_output;
  56     struct pa_memblockq *memblockq;
  57     size_t fragment_size;
  58 };
  59
  60 struct playback_stream {
  61     int type;
  62     struct connection *connection;
  63     uint32_t index;
  64     struct pa_sink_input *sink_input;
  65     struct pa_memblockq *memblockq;
  66     size_t requested_bytes;
  67     int drain_request;
  68     uint32_t drain_tag;
  69 };
  70
  71 struct upload_stream {
  72     int type;
  73     struct connection *connection;
  74     uint32_t index;
  75     struct pa_memchunk memchunk;
  76     size_t length;
  77     char *name;
  78     struct pa_sample_spec sample_spec;
  79 };
  80
  81 struct output_stream {
  82     int type;
  83 };
  84
  85 enum {
  86     UPLOAD_STREAM,
  87     PLAYBACK_STREAM
  88 };
  89
  90 struct connection {
  91     int authorized;
  92     struct pa_protocol_native *protocol;
  93     struct pa_client *client;
  94     struct pa_pstream *pstream;
  95     struct pa_pdispatch *pdispatch;
  96     struct pa_idxset *record_streams, *output_streams;
  97     uint32_t rrobin_index;
  98     struct pa_subscription *subscription;
  99 };
 100
 101 struct pa_protocol_native {
 102     struct pa_module *module;
 103     int public;
 104     struct pa_core *core;
 105     struct pa_socket_server *server;
 106     struct pa_idxset *connections;
 107     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
 108 };
 109
 110 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk);
 111 static void sink_input_drop_cb(struct pa_sink_input *i, const struct pa_memchunk *chunk, size_t length);
 112 static void sink_input_kill_cb(struct pa_sink_input *i);
 113 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i);
 114
 115 static void request_bytes(struct playback_stream*s);
 116
 117 static void source_output_kill_cb(struct pa_source_output *o);
 118 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk);
 119
 120 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 121 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 122 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 123 static void command_create_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 124 static void command_delete_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 125 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 126 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 127 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 128 static void command_stat(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 129 static void command_get_playback_latency(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 130 static void command_create_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 131 static void command_finish_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 132 static void command_play_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 133 static void command_remove_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 134 static void command_get_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 135 static void command_get_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 136 static void command_get_server_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 137 static void command_subscribe(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 138 static void command_set_volume(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 139 static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 140 static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 141
 142 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
 143     [PA_COMMAND_ERROR] = { NULL },
 144     [PA_COMMAND_TIMEOUT] = { NULL },
 145     [PA_COMMAND_REPLY] = { NULL },
 146     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { command_create_playback_stream },
 147     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { command_delete_stream },
 148     [PA_COMMAND_DRAIN_PLAYBACK_STREAM] = { command_drain_playback_stream },
 149     [PA_COMMAND_CREATE_RECORD_STREAM] = { command_create_record_stream },
 150     [PA_COMMAND_DELETE_RECORD_STREAM] = { command_delete_stream },
 151     [PA_COMMAND_AUTH] = { command_auth },
 152     [PA_COMMAND_REQUEST] = { NULL },
 153     [PA_COMMAND_EXIT] = { command_exit },
 154     [PA_COMMAND_SET_NAME] = { command_set_name },
 155     [PA_COMMAND_LOOKUP_SINK] = { command_lookup },
 156     [PA_COMMAND_LOOKUP_SOURCE] = { command_lookup },
 157     [PA_COMMAND_STAT] = { command_stat },
 158     [PA_COMMAND_GET_PLAYBACK_LATENCY] = { command_get_playback_latency },
 159     [PA_COMMAND_CREATE_UPLOAD_STREAM] = { command_create_upload_stream },
 160     [PA_COMMAND_DELETE_UPLOAD_STREAM] = { command_delete_stream },
 161     [PA_COMMAND_FINISH_UPLOAD_STREAM] = { command_finish_upload_stream },
 162     [PA_COMMAND_PLAY_SAMPLE] = { command_play_sample },
 163     [PA_COMMAND_REMOVE_SAMPLE] = { command_remove_sample },
 164     [PA_COMMAND_GET_SINK_INFO] = { command_get_info },
 165     [PA_COMMAND_GET_SOURCE_INFO] = { command_get_info },
 166     [PA_COMMAND_GET_CLIENT_INFO] = { command_get_info },
 167     [PA_COMMAND_GET_MODULE_INFO] = { command_get_info },
 168     [PA_COMMAND_GET_SINK_INPUT_INFO] = { command_get_info },
 169     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO] = { command_get_info },
 170     [PA_COMMAND_GET_SAMPLE_INFO] = { command_get_info },
 171     [PA_COMMAND_GET_SINK_INFO_LIST] = { command_get_info_list },
 172     [PA_COMMAND_GET_SOURCE_INFO_LIST] = { command_get_info_list },
 173     [PA_COMMAND_GET_MODULE_INFO_LIST] = { command_get_info_list },
 174     [PA_COMMAND_GET_CLIENT_INFO_LIST] = { command_get_info_list },
 175     [PA_COMMAND_GET_SINK_INPUT_INFO_LIST] = { command_get_info_list },
 176     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST] = { command_get_info_list },
 177     [PA_COMMAND_GET_SAMPLE_INFO_LIST] = { command_get_info_list },
 178     [PA_COMMAND_GET_SERVER_INFO] = { command_get_server_info },
 179     [PA_COMMAND_SUBSCRIBE] = { command_subscribe },
 180     [PA_COMMAND_SET_SINK_VOLUME] = { command_set_volume },
 181     [PA_COMMAND_SET_SINK_INPUT_VOLUME] = { command_set_volume },
 182     [PA_COMMAND_CORK_PLAYBACK_STREAM] = { command_cork_playback_stream },
 183     [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
 184     [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
 185 };
 186
 187 /* structure management */
 188
 189 static struct upload_stream* upload_stream_new(struct connection *c, const struct pa_sample_spec *ss, const char *name, size_t length) {
 190     struct upload_stream *s;
 191     assert(c && ss && name && length);
 192
 193     s = pa_xmalloc(sizeof(struct upload_stream));
 194     s->type = UPLOAD_STREAM;
 195     s->connection = c;
 196     s->sample_spec = *ss;
 197     s->name = pa_xstrdup(name);
 198
 199     s->memchunk.memblock = NULL;
 200     s->memchunk.index = 0;
 201     s->memchunk.length = 0;
 202
 203     s->length = length;
 204
 205     pa_idxset_put(c->output_streams, s, &s->index);
 206     return s;
 207 }
 208
 209 static void upload_stream_free(struct upload_stream *o) {
 210     assert(o && o->connection);
 211
 212     pa_idxset_remove_by_data(o->connection->output_streams, o, NULL);
 213
 214     pa_xfree(o->name);
 215
 216     if (o->memchunk.memblock)
 217         pa_memblock_unref(o->memchunk.memblock);
 218
 219     pa_xfree(o);
 220 }
 221
 222 static struct record_stream* record_stream_new(struct connection *c, struct pa_source *source, const struct pa_sample_spec *ss, const char *name, size_t maxlength, size_t fragment_size) {
 223     struct record_stream *s;
 224     struct pa_source_output *source_output;
 225     size_t base;
 226     assert(c && source && ss && name && maxlength);
 227
 228     if (!(source_output = pa_source_output_new(source, name, ss)))
 229         return NULL;
 230
 231     s = pa_xmalloc(sizeof(struct record_stream));
 232     s->connection = c;
 233     s->source_output = source_output;
 234     s->source_output->push = source_output_push_cb;
 235     s->source_output->kill = source_output_kill_cb;
 236     s->source_output->userdata = s;
 237     s->source_output->owner = c->protocol->module;
 238     s->source_output->client = c->client;
 239
 240     s->memblockq = pa_memblockq_new(maxlength, 0, base = pa_frame_size(ss), 0, 0, c->protocol->core->memblock_stat);
 241     assert(s->memblockq);
 242
 243     s->fragment_size = (fragment_size/base)*base;
 244     if (!s->fragment_size)
 245         s->fragment_size = base;
 246
 247     pa_idxset_put(c->record_streams, s, &s->index);
 248     return s;
 249 }
 250
 251 static void record_stream_free(struct record_stream* r) {
 252     assert(r && r->connection);
 253
 254     pa_idxset_remove_by_data(r->connection->record_streams, r, NULL);
 255     pa_source_output_free(r->source_output);
 256     pa_memblockq_free(r->memblockq);
 257     pa_xfree(r);
 258 }
 259
 260 static struct playback_stream* playback_stream_new(struct connection *c, struct pa_sink *sink, const struct pa_sample_spec *ss, const char *name,
 261                                                    size_t maxlength,
 262                                                    size_t tlength,
 263                                                    size_t prebuf,
 264                                                    size_t minreq) {
 265     struct playback_stream *s;
 266     struct pa_sink_input *sink_input;
 267     assert(c && sink && ss && name && maxlength);
 268
 269     if (!(sink_input = pa_sink_input_new(sink, name, ss)))
 270         return NULL;
 271
 272     s = pa_xmalloc(sizeof(struct playback_stream));
 273     s->type = PLAYBACK_STREAM;
 274     s->connection = c;
 275     s->sink_input = sink_input;
 276
 277     s->sink_input->peek = sink_input_peek_cb;
 278     s->sink_input->drop = sink_input_drop_cb;
 279     s->sink_input->kill = sink_input_kill_cb;
 280     s->sink_input->get_latency = sink_input_get_latency_cb;
 281     s->sink_input->userdata = s;
 282     s->sink_input->owner = c->protocol->module;
 283     s->sink_input->client = c->client;
 284
 285     s->memblockq = pa_memblockq_new(maxlength, tlength, pa_frame_size(ss), prebuf, minreq, c->protocol->core->memblock_stat);
 286     assert(s->memblockq);
 287
 288     s->requested_bytes = 0;
 289     s->drain_request = 0;
 290
 291     pa_idxset_put(c->output_streams, s, &s->index);
 292     return s;
 293 }
 294
 295 static void playback_stream_free(struct playback_stream* p) {
 296     assert(p && p->connection);
 297
 298     if (p->drain_request)
 299         pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERROR_NOENTITY);
 300
 301     pa_idxset_remove_by_data(p->connection->output_streams, p, NULL);
 302     pa_sink_input_free(p->sink_input);
 303     pa_memblockq_free(p->memblockq);
 304     pa_xfree(p);
 305 }
 306
 307 static void connection_free(struct connection *c) {
 308     struct record_stream *r;
 309     struct output_stream *o;
 310     assert(c && c->protocol);
 311
 312     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 313     while ((r = pa_idxset_first(c->record_streams, NULL)))
 314         record_stream_free(r);
 315     pa_idxset_free(c->record_streams, NULL, NULL);
 316
 317     while ((o = pa_idxset_first(c->output_streams, NULL)))
 318         if (o->type == PLAYBACK_STREAM)
 319             playback_stream_free((struct playback_stream*) o);
 320         else
 321             upload_stream_free((struct upload_stream*) o);
 322     pa_idxset_free(c->output_streams, NULL, NULL);
 323
 324     pa_pdispatch_unref(c->pdispatch);
 325     pa_pstream_close(c->pstream);
 326     pa_pstream_unref(c->pstream);
 327     pa_client_free(c->client);
 328
 329     if (c->subscription)
 330         pa_subscription_free(c->subscription);
 331
 332     pa_xfree(c);
 333 }
 334
 335 static void request_bytes(struct playback_stream *s) {
 336     struct pa_tagstruct *t;
 337     size_t l;
 338     assert(s);
 339
 340     if (!(l = pa_memblockq_missing(s->memblockq)))
 341         return;
 342
 343     if (l <= s->requested_bytes)
 344         return;
 345
 346     l -= s->requested_bytes;
 347
 348     if (l < pa_memblockq_get_minreq(s->memblockq))
 349         return;
 350
 351     s->requested_bytes += l;
 352
 353     t = pa_tagstruct_new(NULL, 0);
 354     assert(t);
 355     pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 356     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 357     pa_tagstruct_putu32(t, s->index);
 358     pa_tagstruct_putu32(t, l);
 359     pa_pstream_send_tagstruct(s->connection->pstream, t);
 360
 361     /*pa_log(__FILE__": Requesting %u bytes\n", l);*/
 362 }
 363
 364 static void send_memblock(struct connection *c) {
 365     uint32_t start;
 366     struct record_stream *r;
 367
 368     start = PA_IDXSET_INVALID;
 369     for (;;) {
 370         struct pa_memchunk chunk;
 371
 372         if (!(r = pa_idxset_rrobin(c->record_streams, &c->rrobin_index)))
 373             return;
 374
 375         if (start == PA_IDXSET_INVALID)
 376             start = c->rrobin_index;
 377         else if (start == c->rrobin_index)
 378             return;
 379
 380         if (pa_memblockq_peek(r->memblockq,  &chunk) >= 0) {
 381             struct pa_memchunk schunk = chunk;
 382
 383             if (schunk.length > r->fragment_size)
 384                 schunk.length = r->fragment_size;
 385
 386             pa_pstream_send_memblock(c->pstream, r->index, 0, &schunk);
 387             pa_memblockq_drop(r->memblockq, &chunk, schunk.length);
 388             pa_memblock_unref(schunk.memblock);
 389
 390             return;
 391         }
 392     }
 393 }
 394
 395 static void send_playback_stream_killed(struct playback_stream *p) {
 396     struct pa_tagstruct *t;
 397     assert(p);
 398
 399     t = pa_tagstruct_new(NULL, 0);
 400     assert(t);
 401     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_KILLED);
 402     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 403     pa_tagstruct_putu32(t, p->index);
 404     pa_pstream_send_tagstruct(p->connection->pstream, t);
 405 }
 406
 407 static void send_record_stream_killed(struct record_stream *r) {
 408     struct pa_tagstruct *t;
 409     assert(r);
 410
 411     t = pa_tagstruct_new(NULL, 0);
 412     assert(t);
 413     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_KILLED);
 414     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 415     pa_tagstruct_putu32(t, r->index);
 416     pa_pstream_send_tagstruct(r->connection->pstream, t);
 417 }
 418
 419
 420 /*** sinkinput callbacks ***/
 421
 422 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk) {
 423     struct playback_stream *s;
 424     assert(i && i->userdata && chunk);
 425     s = i->userdata;
 426
 427     if (pa_memblockq_peek(s->memblockq, chunk) < 0)
 428         return -1;
 429
 430     return 0;
 431 }
 432
 433 static void sink_input_drop_cb(struct pa_sink_input *i, const struct pa_memchunk *chunk, size_t length) {
 434     struct playback_stream *s;
 435     assert(i && i->userdata && length);
 436     s = i->userdata;
 437
 438     pa_memblockq_drop(s->memblockq, chunk, length);
 439     request_bytes(s);
 440
 441     if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) {
 442         pa_pstream_send_simple_ack(s->connection->pstream, s->drain_tag);
 443         s->drain_request = 0;
 444     }
 445
 446     /*pa_log(__FILE__": after_drop: %u\n", pa_memblockq_get_length(s->memblockq));*/
 447 }
 448
 449 static void sink_input_kill_cb(struct pa_sink_input *i) {
 450     assert(i && i->userdata);
 451     send_playback_stream_killed((struct playback_stream *) i->userdata);
 452     playback_stream_free((struct playback_stream *) i->userdata);
 453 }
 454
 455 static uint32_t sink_input_get_latency_cb(struct pa_sink_input *i) {
 456     struct playback_stream *s;
 457     assert(i && i->userdata);
 458     s = i->userdata;
 459
 460     /*pa_log(__FILE__": get_latency: %u\n", pa_memblockq_get_length(s->memblockq));*/
 461
 462     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
 463 }
 464
 465 /*** source_output callbacks ***/
 466
 467 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk) {
 468     struct record_stream *s;
 469     assert(o && o->userdata && chunk);
 470     s = o->userdata;
 471
 472     pa_memblockq_push_align(s->memblockq, chunk, 0);
 473     if (!pa_pstream_is_pending(s->connection->pstream))
 474         send_memblock(s->connection);
 475 }
 476
 477 static void source_output_kill_cb(struct pa_source_output *o) {
 478     assert(o && o->userdata);
 479     send_record_stream_killed((struct record_stream *) o->userdata);
 480     record_stream_free((struct record_stream *) o->userdata);
 481 }
 482
 483 /*** pdispatch callbacks ***/
 484
 485 static void protocol_error(struct connection *c) {
 486     pa_log(__FILE__": protocol error, kicking client\n");
 487     connection_free(c);
 488 }
 489
 490 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 491     struct connection *c = userdata;
 492     struct playback_stream *s;
 493     size_t maxlength, tlength, prebuf, minreq;
 494     uint32_t sink_index;
 495     const char *name, *sink_name;
 496     struct pa_sample_spec ss;
 497     struct pa_tagstruct *reply;
 498     struct pa_sink *sink;
 499     assert(c && t && c->protocol && c->protocol->core);
 500
 501     if (pa_tagstruct_gets(t, &name) < 0 ||
 502         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 503         pa_tagstruct_getu32(t, &sink_index) < 0 ||
 504         pa_tagstruct_gets(t, &sink_name) < 0 ||
 505         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 506         pa_tagstruct_getu32(t, &tlength) < 0 ||
 507         pa_tagstruct_getu32(t, &prebuf) < 0 ||
 508         pa_tagstruct_getu32(t, &minreq) < 0 ||
 509         !pa_tagstruct_eof(t)) {
 510         protocol_error(c);
 511         return;
 512     }
 513
 514     if (!c->authorized) {
 515         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 516         return;
 517     }
 518
 519     if (sink_index != (uint32_t) -1)
 520         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
 521     else
 522         sink = pa_namereg_get(c->protocol->core, *sink_name ? sink_name : NULL, PA_NAMEREG_SINK, 1);
 523
 524     if (!sink) {
 525         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 526         return;
 527     }
 528
 529     if (!(s = playback_stream_new(c, sink, &ss, name, maxlength, tlength, prebuf, minreq))) {
 530         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 531         return;
 532     }
 533
 534     reply = pa_tagstruct_new(NULL, 0);
 535     assert(reply);
 536     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 537     pa_tagstruct_putu32(reply, tag);
 538     pa_tagstruct_putu32(reply, s->index);
 539     assert(s->sink_input);
 540     pa_tagstruct_putu32(reply, s->sink_input->index);
 541     pa_pstream_send_tagstruct(c->pstream, reply);
 542     request_bytes(s);
 543 }
 544
 545 static void command_delete_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 546     struct connection *c = userdata;
 547     uint32_t channel;
 548     assert(c && t);
 549
 550     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 551         !pa_tagstruct_eof(t)) {
 552         protocol_error(c);
 553         return;
 554     }
 555
 556     if (!c->authorized) {
 557         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 558         return;
 559     }
 560
 561     if (command == PA_COMMAND_DELETE_PLAYBACK_STREAM) {
 562         struct playback_stream *s;
 563         if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != PLAYBACK_STREAM)) {
 564             pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 565             return;
 566         }
 567
 568         playback_stream_free(s);
 569     } else if (command == PA_COMMAND_DELETE_RECORD_STREAM) {
 570         struct record_stream *s;
 571         if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
 572             pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 573             return;
 574         }
 575
 576         record_stream_free(s);
 577     } else {
 578         struct upload_stream *s;
 579         assert(command == PA_COMMAND_DELETE_UPLOAD_STREAM);
 580         if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) {
 581             pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 582             return;
 583         }
 584
 585         upload_stream_free(s);
 586     }
 587
 588     pa_pstream_send_simple_ack(c->pstream, tag);
 589 }
 590
 591 static void command_create_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 592     struct connection *c = userdata;
 593     struct record_stream *s;
 594     size_t maxlength, fragment_size;
 595     uint32_t source_index;
 596     const char *name, *source_name;
 597     struct pa_sample_spec ss;
 598     struct pa_tagstruct *reply;
 599     struct pa_source *source;
 600     assert(c && t && c->protocol && c->protocol->core);
 601
 602     if (pa_tagstruct_gets(t, &name) < 0 ||
 603         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 604         pa_tagstruct_getu32(t, &source_index) < 0 ||
 605         pa_tagstruct_gets(t, &source_name) < 0 ||
 606         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 607         pa_tagstruct_getu32(t, &fragment_size) < 0 ||
 608         !pa_tagstruct_eof(t)) {
 609         protocol_error(c);
 610         return;
 611     }
 612
 613     if (!c->authorized) {
 614         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 615         return;
 616     }
 617
 618     if (source_index != (uint32_t) -1)
 619         source = pa_idxset_get_by_index(c->protocol->core->sources, source_index);
 620     else
 621         source = pa_namereg_get(c->protocol->core, *source_name ? source_name : NULL, PA_NAMEREG_SOURCE, 1);
 622
 623     if (!source) {
 624         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 625         return;
 626     }
 627
 628     if (!(s = record_stream_new(c, source, &ss, name, maxlength, fragment_size))) {
 629         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 630         return;
 631     }
 632
 633     reply = pa_tagstruct_new(NULL, 0);
 634     assert(reply);
 635     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 636     pa_tagstruct_putu32(reply, tag);
 637     pa_tagstruct_putu32(reply, s->index);
 638     assert(s->source_output);
 639     pa_tagstruct_putu32(reply, s->source_output->index);
 640     pa_pstream_send_tagstruct(c->pstream, reply);
 641 }
 642
 643 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 644     struct connection *c = userdata;
 645     assert(c && t);
 646
 647     if (!pa_tagstruct_eof(t)) {
 648         protocol_error(c);
 649         return;
 650     }
 651
 652     if (!c->authorized) {
 653         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 654         return;
 655     }
 656
 657     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop);
 658     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
 659     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
 660     return;
 661 }
 662
 663 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 664     struct connection *c = userdata;
 665     const void*cookie;
 666     assert(c && t);
 667
 668     if (pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
 669         !pa_tagstruct_eof(t)) {
 670         protocol_error(c);
 671         return;
 672     }
 673
 674     if (memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) != 0) {
 675         pa_log(__FILE__": Denied access to client with invalid authorization key.\n");
 676         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 677         return;
 678     }
 679
 680     c->authorized = 1;
 681     pa_pstream_send_simple_ack(c->pstream, tag);
 682     return;
 683 }
 684
 685 static void command_set_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 686     struct connection *c = userdata;
 687     const char *name;
 688     assert(c && t);
 689
 690     if (pa_tagstruct_gets(t, &name) < 0 ||
 691         !pa_tagstruct_eof(t)) {
 692         protocol_error(c);
 693         return;
 694     }
 695
 696     pa_client_rename(c->client, name);
 697     pa_pstream_send_simple_ack(c->pstream, tag);
 698     return;
 699 }
 700
 701 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 702     struct connection *c = userdata;
 703     const char *name;
 704     uint32_t index = PA_IDXSET_INVALID;
 705     assert(c && t);
 706
 707     if (pa_tagstruct_gets(t, &name) < 0 ||
 708         !pa_tagstruct_eof(t)) {
 709         protocol_error(c);
 710         return;
 711     }
 712
 713     if (!c->authorized) {
 714         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 715         return;
 716     }
 717
 718     if (command == PA_COMMAND_LOOKUP_SINK) {
 719         struct pa_sink *sink;
 720         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1)))
 721             index = sink->index;
 722     } else {
 723         struct pa_source *source;
 724         assert(command == PA_COMMAND_LOOKUP_SOURCE);
 725         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1)))
 726             index = source->index;
 727     }
 728
 729     if (index == PA_IDXSET_INVALID)
 730         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 731     else {
 732         struct pa_tagstruct *reply;
 733         reply = pa_tagstruct_new(NULL, 0);
 734         assert(reply);
 735         pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 736         pa_tagstruct_putu32(reply, tag);
 737         pa_tagstruct_putu32(reply, index);
 738         pa_pstream_send_tagstruct(c->pstream, reply);
 739     }
 740 }
 741
 742 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 743     struct connection *c = userdata;
 744     uint32_t index;
 745     struct playback_stream *s;
 746     assert(c && t);
 747
 748     if (pa_tagstruct_getu32(t, &index) < 0 ||
 749         !pa_tagstruct_eof(t)) {
 750         protocol_error(c);
 751         return;
 752     }
 753
 754     if (!c->authorized) {
 755         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 756         return;
 757     }
 758
 759     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
 760         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 761         return;
 762     }
 763
 764     s->drain_request = 0;
 765
 766     pa_memblockq_prebuf_disable(s->memblockq);
 767
 768     if (!pa_memblockq_is_readable(s->memblockq)) {
 769         pa_pstream_send_simple_ack(c->pstream, tag);
 770     } else {
 771         s->drain_request = 1;
 772         s->drain_tag = tag;
 773
 774         pa_sink_notify(s->sink_input->sink);
 775     }
 776 }
 777
 778 static void command_stat(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 779     struct connection *c = userdata;
 780     struct pa_tagstruct *reply;
 781     assert(c && t);
 782
 783     if (!pa_tagstruct_eof(t)) {
 784         protocol_error(c);
 785         return;
 786     }
 787
 788     if (!c->authorized) {
 789         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 790         return;
 791     }
 792
 793     reply = pa_tagstruct_new(NULL, 0);
 794     assert(reply);
 795     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 796     pa_tagstruct_putu32(reply, tag);
 797     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->total);
 798     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->total_size);
 799     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->allocated);
 800     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->allocated_size);
 801     pa_tagstruct_putu32(reply, pa_scache_total_size(c->protocol->core));
 802     pa_pstream_send_tagstruct(c->pstream, reply);
 803 }
 804
 805 static void command_get_playback_latency(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 806     struct connection *c = userdata;
 807     struct pa_tagstruct *reply;
 808     struct playback_stream *s;
 809     uint32_t index;
 810     assert(c && t);
 811
 812     if (pa_tagstruct_getu32(t, &index) < 0 ||
 813         !pa_tagstruct_eof(t)) {
 814         protocol_error(c);
 815         return;
 816     }
 817
 818     if (!c->authorized) {
 819         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 820         return;
 821     }
 822
 823     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
 824         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 825         return;
 826     }
 827
 828     reply = pa_tagstruct_new(NULL, 0);
 829     assert(reply);
 830     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 831     pa_tagstruct_putu32(reply, tag);
 832     pa_tagstruct_putu32(reply, pa_sink_input_get_latency(s->sink_input));
 833     pa_tagstruct_putu32(reply, pa_sink_get_latency(s->sink_input->sink));
 834     pa_tagstruct_put_boolean(reply, pa_memblockq_is_readable(s->memblockq));
 835     pa_tagstruct_putu32(reply, pa_memblockq_get_length(s->memblockq));
 836     pa_pstream_send_tagstruct(c->pstream, reply);
 837 }
 838
 839 static void command_create_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 840     struct connection *c = userdata;
 841     struct upload_stream *s;
 842     size_t length;
 843     const char *name;
 844     struct pa_sample_spec ss;
 845     struct pa_tagstruct *reply;
 846     assert(c && t && c->protocol && c->protocol->core);
 847
 848     if (pa_tagstruct_gets(t, &name) < 0 ||
 849         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 850         pa_tagstruct_getu32(t, &length) < 0 ||
 851         !pa_tagstruct_eof(t)) {
 852         protocol_error(c);
 853         return;
 854     }
 855
 856     if (!c->authorized) {
 857         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 858         return;
 859     }
 860
 861     if ((length % pa_frame_size(&ss)) != 0 || length <= 0 || !*name) {
 862         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 863         return;
 864     }
 865
 866     if (!(s = upload_stream_new(c, &ss, name, length))) {
 867         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 868         return;
 869     }
 870
 871     reply = pa_tagstruct_new(NULL, 0);
 872     assert(reply);
 873     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 874     pa_tagstruct_putu32(reply, tag);
 875     pa_tagstruct_putu32(reply, s->index);
 876     pa_pstream_send_tagstruct(c->pstream, reply);
 877
 878     reply = pa_tagstruct_new(NULL, 0);
 879     assert(reply);
 880     pa_tagstruct_putu32(reply, PA_COMMAND_REQUEST);
 881     pa_tagstruct_putu32(reply, (uint32_t) -1); /* tag */
 882     pa_tagstruct_putu32(reply, s->index);
 883     pa_tagstruct_putu32(reply, length);
 884     pa_pstream_send_tagstruct(c->pstream, reply);
 885 }
 886
 887 static void command_finish_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 888     struct connection *c = userdata;
 889     uint32_t channel;
 890     struct upload_stream *s;
 891     uint32_t index;
 892     assert(c && t);
 893
 894     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 895         !pa_tagstruct_eof(t)) {
 896         protocol_error(c);
 897         return;
 898     }
 899
 900     if (!c->authorized) {
 901         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 902         return;
 903     }
 904
 905     if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) {
 906         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 907         return;
 908     }
 909
 910     pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->memchunk, &index);
 911     pa_pstream_send_simple_ack(c->pstream, tag);
 912     upload_stream_free(s);
 913 }
 914
 915 static void command_play_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 916     struct connection *c = userdata;
 917     uint32_t sink_index, volume;
 918     struct pa_sink *sink;
 919     const char *name, *sink_name;
 920     assert(c && t);
 921
 922     if (pa_tagstruct_getu32(t, &sink_index) < 0 ||
 923         pa_tagstruct_gets(t, &sink_name) < 0 ||
 924         pa_tagstruct_getu32(t, &volume) < 0 ||
 925         pa_tagstruct_gets(t, &name) < 0 ||
 926         !pa_tagstruct_eof(t)) {
 927         protocol_error(c);
 928         return;
 929     }
 930
 931     if (!c->authorized) {
 932         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 933         return;
 934     }
 935
 936     if (sink_index != (uint32_t) -1)
 937         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
 938     else
 939         sink = pa_namereg_get(c->protocol->core, *sink_name ? sink_name : NULL, PA_NAMEREG_SINK, 1);
 940
 941     if (!sink) {
 942         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 943         return;
 944     }
 945
 946     if (pa_scache_play_item(c->protocol->core, name, sink, volume) < 0) {
 947         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 948         return;
 949     }
 950
 951     pa_pstream_send_simple_ack(c->pstream, tag);
 952 }
 953
 954 static void command_remove_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 955     struct connection *c = userdata;
 956     const char *name;
 957     assert(c && t);
 958
 959     if (pa_tagstruct_gets(t, &name) < 0 ||
 960         !pa_tagstruct_eof(t)) {
 961         protocol_error(c);
 962         return;
 963     }
 964
 965     if (!c->authorized) {
 966         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 967         return;
 968     }
 969
 970     if (pa_scache_remove_item(c->protocol->core, name) < 0) {
 971         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 972         return;
 973     }
 974
 975     pa_pstream_send_simple_ack(c->pstream, tag);
 976 }
 977
 978 static void sink_fill_tagstruct(struct pa_tagstruct *t, struct pa_sink *sink) {
 979     assert(t && sink);
 980     pa_tagstruct_putu32(t, sink->index);
 981     pa_tagstruct_puts(t, sink->name);
 982     pa_tagstruct_puts(t, sink->description ? sink->description : "");
 983     pa_tagstruct_put_sample_spec(t, &sink->sample_spec);
 984     pa_tagstruct_putu32(t, sink->owner ? sink->owner->index : (uint32_t) -1);
 985     pa_tagstruct_putu32(t, sink->volume);
 986     pa_tagstruct_putu32(t, sink->monitor_source->index);
 987     pa_tagstruct_puts(t, sink->monitor_source->name);
 988     pa_tagstruct_putu32(t, pa_sink_get_latency(sink));
 989 }
 990
 991 static void source_fill_tagstruct(struct pa_tagstruct *t, struct pa_source *source) {
 992     assert(t && source);
 993     pa_tagstruct_putu32(t, source->index);
 994     pa_tagstruct_puts(t, source->name);
 995     pa_tagstruct_puts(t, source->description ? source->description : "");
 996     pa_tagstruct_put_sample_spec(t, &source->sample_spec);
 997     pa_tagstruct_putu32(t, source->owner ? source->owner->index : (uint32_t) -1);
 998     pa_tagstruct_putu32(t, source->monitor_of ? source->monitor_of->index : (uint32_t) -1);
 999     pa_tagstruct_puts(t, source->monitor_of ? source->monitor_of->name : "");
1000 }
1001
1002 static void client_fill_tagstruct(struct pa_tagstruct *t, struct pa_client *client) {
1003     assert(t && client);
1004     pa_tagstruct_putu32(t, client->index);
1005     pa_tagstruct_puts(t, client->name);
1006     pa_tagstruct_puts(t, client->protocol_name);
1007     pa_tagstruct_putu32(t, client->owner ? client->owner->index : (uint32_t) -1);
1008 }
1009
1010 static void module_fill_tagstruct(struct pa_tagstruct *t, struct pa_module *module) {
1011     assert(t && module);
1012     pa_tagstruct_putu32(t, module->index);
1013     pa_tagstruct_puts(t, module->name);
1014     pa_tagstruct_puts(t, module->argument ? module->argument : "");
1015     pa_tagstruct_putu32(t, module->n_used);
1016     pa_tagstruct_put_boolean(t, module->auto_unload);
1017 }
1018
1019 static void sink_input_fill_tagstruct(struct pa_tagstruct *t, struct pa_sink_input *s) {
1020     assert(t && s);
1021     pa_tagstruct_putu32(t, s->index);
1022     pa_tagstruct_puts(t, s->name ? s->name : "");
1023     pa_tagstruct_putu32(t, s->owner ? s->owner->index : (uint32_t) -1);
1024     pa_tagstruct_putu32(t, s->client ? s->client->index : (uint32_t) -1);
1025     pa_tagstruct_putu32(t, s->sink->index);
1026     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1027     pa_tagstruct_putu32(t, s->volume);
1028     pa_tagstruct_putu32(t, pa_sink_input_get_latency(s));
1029     pa_tagstruct_putu32(t, pa_sink_get_latency(s->sink));
1030 }
1031
1032 static void source_output_fill_tagstruct(struct pa_tagstruct *t, struct pa_source_output *s) {
1033     assert(t && s);
1034     pa_tagstruct_putu32(t, s->index);
1035     pa_tagstruct_puts(t, s->name ? s->name : "");
1036     pa_tagstruct_putu32(t, s->owner ? s->owner->index : (uint32_t) -1);
1037     pa_tagstruct_putu32(t, s->client ? s->client->index : (uint32_t) -1);
1038     pa_tagstruct_putu32(t, s->source->index);
1039     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1040 }
1041
1042 static void scache_fill_tagstruct(struct pa_tagstruct *t, struct pa_scache_entry *e) {
1043     assert(t && e);
1044     pa_tagstruct_putu32(t, e->index);
1045     pa_tagstruct_puts(t, e->name);
1046     pa_tagstruct_putu32(t, e->volume);
1047     pa_tagstruct_putu32(t, pa_bytes_to_usec(e->memchunk.length, &e->sample_spec));
1048     pa_tagstruct_put_sample_spec(t, &e->sample_spec);
1049     pa_tagstruct_putu32(t, e->memchunk.length);
1050 }
1051
1052 static void command_get_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1053     struct connection *c = userdata;
1054     uint32_t index;
1055     struct pa_sink *sink = NULL;
1056     struct pa_source *source = NULL;
1057     struct pa_client *client = NULL;
1058     struct pa_module *module = NULL;
1059     struct pa_sink_input *si = NULL;
1060     struct pa_source_output *so = NULL;
1061     struct pa_scache_entry *sce = NULL;
1062     const char *name;
1063     struct pa_tagstruct *reply;
1064     assert(c && t);
1065
1066
1067     if (pa_tagstruct_getu32(t, &index) < 0 ||
1068         (command != PA_COMMAND_GET_CLIENT_INFO &&
1069          command != PA_COMMAND_GET_MODULE_INFO &&
1070          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
1071          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
1072          pa_tagstruct_gets(t, &name) < 0) ||
1073         !pa_tagstruct_eof(t)) {
1074         protocol_error(c);
1075         return;
1076     }
1077
1078     if (!c->authorized) {
1079         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1080         return;
1081     }
1082
1083     if (command == PA_COMMAND_GET_SINK_INFO) {
1084         if (index != (uint32_t) -1)
1085             sink = pa_idxset_get_by_index(c->protocol->core->sinks, index);
1086         else
1087             sink = pa_namereg_get(c->protocol->core, *name ? name : NULL, PA_NAMEREG_SINK, 1);
1088     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
1089         if (index != (uint32_t) -1)
1090             source = pa_idxset_get_by_index(c->protocol->core->sources, index);
1091         else
1092             source = pa_namereg_get(c->protocol->core, *name ? name : NULL, PA_NAMEREG_SOURCE, 1);
1093     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
1094         client = pa_idxset_get_by_index(c->protocol->core->clients, index);
1095     else if (command == PA_COMMAND_GET_MODULE_INFO)
1096         module = pa_idxset_get_by_index(c->protocol->core->modules, index);
1097     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
1098         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index);
1099     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
1100         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, index);
1101     else {
1102         assert(command == PA_COMMAND_GET_SAMPLE_INFO && name);
1103         if (index != (uint32_t) -1)
1104             sce = pa_idxset_get_by_index(c->protocol->core->scache, index);
1105         else
1106             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE, 0);
1107     }
1108
1109     if (!sink && !source && !client && !module && !si && !so && !sce) {
1110         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1111         return;
1112     }
1113
1114     reply = pa_tagstruct_new(NULL, 0);
1115     assert(reply);
1116     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1117     pa_tagstruct_putu32(reply, tag);
1118     if (sink)
1119         sink_fill_tagstruct(reply, sink);
1120     else if (source)
1121         source_fill_tagstruct(reply, source);
1122     else if (client)
1123         client_fill_tagstruct(reply, client);
1124     else if (module)
1125         module_fill_tagstruct(reply, module);
1126     else if (si)
1127         sink_input_fill_tagstruct(reply, si);
1128     else if (so)
1129         source_output_fill_tagstruct(reply, so);
1130     else
1131         scache_fill_tagstruct(reply, sce);
1132     pa_pstream_send_tagstruct(c->pstream, reply);
1133 }
1134
1135 static void command_get_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1136     struct connection *c = userdata;
1137     struct pa_idxset *i;
1138     uint32_t index;
1139     void *p;
1140     struct pa_tagstruct *reply;
1141     assert(c && t);
1142
1143     if (!pa_tagstruct_eof(t)) {
1144         protocol_error(c);
1145         return;
1146     }
1147
1148     if (!c->authorized) {
1149         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1150         return;
1151     }
1152
1153     reply = pa_tagstruct_new(NULL, 0);
1154     assert(reply);
1155     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1156     pa_tagstruct_putu32(reply, tag);
1157
1158     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
1159         i = c->protocol->core->sinks;
1160     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
1161         i = c->protocol->core->sources;
1162     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
1163         i = c->protocol->core->clients;
1164     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
1165         i = c->protocol->core->modules;
1166     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
1167         i = c->protocol->core->sink_inputs;
1168     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
1169         i = c->protocol->core->source_outputs;
1170     else {
1171         assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
1172         i = c->protocol->core->scache;
1173     }
1174
1175     if (i) {
1176         for (p = pa_idxset_first(i, &index); p; p = pa_idxset_next(i, &index)) {
1177             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
1178                 sink_fill_tagstruct(reply, p);
1179             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
1180                 source_fill_tagstruct(reply, p);
1181             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
1182                 client_fill_tagstruct(reply, p);
1183             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
1184                 module_fill_tagstruct(reply, p);
1185             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
1186                 sink_input_fill_tagstruct(reply, p);
1187             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
1188                 source_output_fill_tagstruct(reply, p);
1189             else {
1190                 assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
1191                 scache_fill_tagstruct(reply, p);
1192             }
1193         }
1194     }
1195
1196     pa_pstream_send_tagstruct(c->pstream, reply);
1197 }
1198
1199 static void command_get_server_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1200     struct connection *c = userdata;
1201     struct pa_tagstruct *reply;
1202     char txt[256];
1203     assert(c && t);
1204
1205     if (!pa_tagstruct_eof(t)) {
1206         protocol_error(c);
1207         return;
1208     }
1209
1210     if (!c->authorized) {
1211         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1212         return;
1213     }
1214
1215     reply = pa_tagstruct_new(NULL, 0);
1216     assert(reply);
1217     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1218     pa_tagstruct_putu32(reply, tag);
1219     pa_tagstruct_puts(reply, PACKAGE_NAME);
1220     pa_tagstruct_puts(reply, PACKAGE_VERSION);
1221     pa_tagstruct_puts(reply, pa_get_user_name(txt, sizeof(txt)));
1222     pa_tagstruct_puts(reply, pa_get_host_name(txt, sizeof(txt)));
1223     pa_tagstruct_put_sample_spec(reply, &c->protocol->core->default_sample_spec);
1224     pa_pstream_send_tagstruct(c->pstream, reply);
1225 }
1226
1227 static void subscription_cb(struct pa_core *core, enum pa_subscription_event_type e, uint32_t index, void *userdata) {
1228     struct pa_tagstruct *t;
1229     struct connection *c = userdata;
1230     assert(c && core);
1231
1232     t = pa_tagstruct_new(NULL, 0);
1233     assert(t);
1234     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
1235     pa_tagstruct_putu32(t, (uint32_t) -1);
1236     pa_tagstruct_putu32(t, e);
1237     pa_tagstruct_putu32(t, index);
1238     pa_pstream_send_tagstruct(c->pstream, t);
1239 }
1240
1241 static void command_subscribe(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1242     struct connection *c = userdata;
1243     enum pa_subscription_mask m;
1244     assert(c && t);
1245
1246     if (pa_tagstruct_getu32(t, &m) < 0 ||
1247         !pa_tagstruct_eof(t)) {
1248         protocol_error(c);
1249         return;
1250     }
1251
1252     if (!c->authorized) {
1253         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1254         return;
1255     }
1256
1257     if (c->subscription)
1258         pa_subscription_free(c->subscription);
1259
1260     if (m != 0) {
1261         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
1262         assert(c->subscription);
1263     } else
1264         c->subscription = NULL;
1265
1266     pa_pstream_send_simple_ack(c->pstream, tag);
1267 }
1268
1269 static void command_set_volume(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1270     struct connection *c = userdata;
1271     uint32_t index, volume;
1272     struct pa_sink *sink = NULL;
1273     struct pa_sink_input *si = NULL;
1274     const char *name = NULL;
1275     assert(c && t);
1276
1277     if (pa_tagstruct_getu32(t, &index) < 0 ||
1278         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
1279         pa_tagstruct_getu32(t, &volume) ||
1280         !pa_tagstruct_eof(t)) {
1281         protocol_error(c);
1282         return;
1283     }
1284
1285     if (!c->authorized) {
1286         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1287         return;
1288     }
1289
1290     if (command == PA_COMMAND_SET_SINK_VOLUME) {
1291         if (index != (uint32_t) -1)
1292             sink = pa_idxset_get_by_index(c->protocol->core->sinks, index);
1293         else
1294             sink = pa_namereg_get(c->protocol->core, *name ? name : NULL, PA_NAMEREG_SINK, 1);
1295     }  else {
1296         assert(command == PA_COMMAND_SET_SINK_INPUT_VOLUME);
1297         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index);
1298     }
1299
1300     if (!si && !sink) {
1301         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1302         return;
1303     }
1304
1305     if (sink)
1306         pa_sink_set_volume(sink, volume);
1307     else if (si)
1308         pa_sink_input_set_volume(si, volume);
1309
1310     pa_pstream_send_simple_ack(c->pstream, tag);
1311 }
1312
1313 static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1314     struct connection *c = userdata;
1315     uint32_t index;
1316     uint32_t b;
1317     struct playback_stream *s;
1318     assert(c && t);
1319
1320     if (pa_tagstruct_getu32(t, &index) < 0 ||
1321         pa_tagstruct_getu32(t, &b) < 0 ||
1322         !pa_tagstruct_eof(t)) {
1323         protocol_error(c);
1324         return;
1325     }
1326
1327     if (!c->authorized) {
1328         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1329         return;
1330     }
1331
1332     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1333         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1334         return;
1335     }
1336
1337     pa_sink_input_cork(s->sink_input, b);
1338     pa_pstream_send_simple_ack(c->pstream, tag);
1339 }
1340
1341 static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1342     struct connection *c = userdata;
1343     uint32_t index;
1344     struct playback_stream *s;
1345     assert(c && t);
1346
1347     if (pa_tagstruct_getu32(t, &index) < 0 ||
1348         !pa_tagstruct_eof(t)) {
1349         protocol_error(c);
1350         return;
1351     }
1352
1353     if (!c->authorized) {
1354         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1355         return;
1356     }
1357
1358     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1359         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1360         return;
1361     }
1362
1363     if (command == PA_COMMAND_TRIGGER_PLAYBACK_STREAM)
1364         pa_memblockq_prebuf_disable(s->memblockq);
1365     else {
1366         assert(command == PA_COMMAND_FLUSH_PLAYBACK_STREAM);
1367         pa_memblockq_flush(s->memblockq);
1368         /*pa_log(__FILE__": flush: %u\n", pa_memblockq_get_length(s->memblockq));*/
1369     }
1370
1371     pa_sink_notify(s->sink_input->sink);
1372     pa_pstream_send_simple_ack(c->pstream, tag);
1373     request_bytes(s);
1374 }
1375
1376 /*** pstream callbacks ***/
1377
1378 static void pstream_packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
1379     struct connection *c = userdata;
1380     assert(p && packet && packet->data && c);
1381
1382     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
1383         pa_log(__FILE__": invalid packet.\n");
1384         connection_free(c);
1385     }
1386 }
1387
1388 static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, uint32_t delta, const struct pa_memchunk *chunk, void *userdata) {
1389     struct connection *c = userdata;
1390     struct output_stream *stream;
1391     assert(p && chunk && userdata);
1392
1393     if (!(stream = pa_idxset_get_by_index(c->output_streams, channel))) {
1394         pa_log(__FILE__": client sent block for invalid stream.\n");
1395         connection_free(c);
1396         return;
1397     }
1398
1399     if (stream->type == PLAYBACK_STREAM) {
1400         struct playback_stream *p = (struct playback_stream*) stream;
1401         if (chunk->length >= p->requested_bytes)
1402             p->requested_bytes = 0;
1403         else
1404             p->requested_bytes -= chunk->length;
1405
1406         pa_memblockq_push_align(p->memblockq, chunk, delta);
1407         assert(p->sink_input);
1408         /*pa_log(__FILE__": after_recv: %u\n", pa_memblockq_get_length(p->memblockq));*/
1409
1410         pa_sink_notify(p->sink_input->sink);
1411         /*pa_log(__FILE__": Recieved %u bytes.\n", chunk->length);*/
1412
1413     } else {
1414         struct upload_stream *u = (struct upload_stream*) stream;
1415         size_t l;
1416         assert(u->type == UPLOAD_STREAM);
1417
1418         if (!u->memchunk.memblock) {
1419             if (u->length == chunk->length) {
1420                 u->memchunk = *chunk;
1421                 pa_memblock_ref(u->memchunk.memblock);
1422                 u->length = 0;
1423             } else {
1424                 u->memchunk.memblock = pa_memblock_new(u->length, c->protocol->core->memblock_stat);
1425                 u->memchunk.index = u->memchunk.length = 0;
1426             }
1427         }
1428
1429         assert(u->memchunk.memblock);
1430
1431         l = u->length;
1432         if (l > chunk->length)
1433             l = chunk->length;
1434
1435         if (l > 0) {
1436             memcpy((uint8_t*) u->memchunk.memblock->data + u->memchunk.index + u->memchunk.length,
1437                    (uint8_t*) chunk->memblock->data+chunk->index, l);
1438             u->memchunk.length += l;
1439             u->length -= l;
1440         }
1441     }
1442 }
1443
1444 static void pstream_die_callback(struct pa_pstream *p, void *userdata) {
1445     struct connection *c = userdata;
1446     assert(p && c);
1447     connection_free(c);
1448
1449 /*    pa_log(__FILE__": connection died.\n");*/
1450 }
1451
1452
1453 static void pstream_drain_callback(struct pa_pstream *p, void *userdata) {
1454     struct connection *c = userdata;
1455     assert(p && c);
1456
1457     send_memblock(c);
1458 }
1459
1460 /*** client callbacks ***/
1461
1462 static void client_kill_cb(struct pa_client *c) {
1463     assert(c && c->userdata);
1464     connection_free(c->userdata);
1465 }
1466
1467 /*** socket server callbacks ***/
1468
1469 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
1470     struct pa_protocol_native *p = userdata;
1471     struct connection *c;
1472     assert(io && p);
1473
1474     c = pa_xmalloc(sizeof(struct connection));
1475     c->authorized = p->public;
1476     c->protocol = p;
1477     assert(p->core);
1478     c->client = pa_client_new(p->core, "NATIVE", "Client");
1479     assert(c->client);
1480     c->client->kill = client_kill_cb;
1481     c->client->userdata = c;
1482     c->client->owner = p->module;
1483
1484     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->memblock_stat);
1485     assert(c->pstream);
1486
1487     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
1488     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
1489     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
1490     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
1491
1492     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
1493     assert(c->pdispatch);
1494
1495     c->record_streams = pa_idxset_new(NULL, NULL);
1496     c->output_streams = pa_idxset_new(NULL, NULL);
1497     assert(c->record_streams && c->output_streams);
1498
1499     c->rrobin_index = PA_IDXSET_INVALID;
1500     c->subscription = NULL;
1501
1502     pa_idxset_put(p->connections, c, NULL);
1503 }
1504
1505 /*** module entry points ***/
1506
1507 static struct pa_protocol_native* protocol_new_internal(struct pa_core *c, struct pa_module *m, struct pa_modargs *ma) {
1508     struct pa_protocol_native *p;
1509     int public;
1510     assert(c && ma);
1511
1512     if (pa_modargs_get_value_boolean(ma, "public", &public) < 0) {
1513         pa_log(__FILE__": public= expects numeric argument.\n");
1514         return NULL;
1515     }
1516
1517     p = pa_xmalloc(sizeof(struct pa_protocol_native));
1518
1519     if (pa_authkey_load_from_home(pa_modargs_get_value(ma, "cookie", PA_NATIVE_COOKIE_FILE), p->auth_cookie, sizeof(p->auth_cookie)) < 0) {
1520         pa_xfree(p);
1521         return NULL;
1522     }
1523
1524     p->module = m;
1525     p->public = public;
1526     p->server = NULL;
1527     p->core = c;
1528     p->connections = pa_idxset_new(NULL, NULL);
1529     assert(p->connections);
1530
1531     return p;
1532 }
1533
1534 struct pa_protocol_native* pa_protocol_native_new(struct pa_core *core, struct pa_socket_server *server, struct pa_module *m, struct pa_modargs *ma) {
1535     struct pa_protocol_native *p;
1536
1537     if (!(p = protocol_new_internal(core, m, ma)))
1538         return NULL;
1539
1540     p->server = server;
1541     pa_socket_server_set_callback(p->server, on_connection, p);
1542
1543     return p;
1544 }
1545
1546 void pa_protocol_native_free(struct pa_protocol_native *p) {
1547     struct connection *c;
1548     assert(p);
1549
1550     while ((c = pa_idxset_first(p->connections, NULL)))
1551         connection_free(c);
1552     pa_idxset_free(p->connections, NULL, NULL);
1553
1554     if (p->server)
1555         pa_socket_server_unref(p->server);
1556
1557     pa_xfree(p);
1558 }
1559
1560 struct pa_protocol_native* pa_protocol_native_new_iochannel(struct pa_core*core, struct pa_iochannel *io, struct pa_module *m, struct pa_modargs *ma) {
1561     struct pa_protocol_native *p;
1562
1563     if (!(p = protocol_new_internal(core, m, ma)))
1564         return NULL;
1565
1566     on_connection(NULL, io, p);
1567
1568     return p;
1569 }