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 #include "autoload.h"
  49 #include "authkey-prop.h"
  50
  51 struct connection;
  52 struct pa_protocol_native;
  53
  54 struct record_stream {
  55     struct connection *connection;
  56     uint32_t index;
  57     struct pa_source_output *source_output;
  58     struct pa_memblockq *memblockq;
  59     size_t fragment_size;
  60 };
  61
  62 struct playback_stream {
  63     int type;
  64     struct connection *connection;
  65     uint32_t index;
  66     struct pa_sink_input *sink_input;
  67     struct pa_memblockq *memblockq;
  68     size_t requested_bytes;
  69     int drain_request;
  70     uint32_t drain_tag;
  71 };
  72
  73 struct upload_stream {
  74     int type;
  75     struct connection *connection;
  76     uint32_t index;
  77     struct pa_memchunk memchunk;
  78     size_t length;
  79     char *name;
  80     struct pa_sample_spec sample_spec;
  81 };
  82
  83 struct output_stream {
  84     int type;
  85 };
  86
  87 enum {
  88     UPLOAD_STREAM,
  89     PLAYBACK_STREAM
  90 };
  91
  92 struct connection {
  93     int authorized;
  94     struct pa_protocol_native *protocol;
  95     struct pa_client *client;
  96     struct pa_pstream *pstream;
  97     struct pa_pdispatch *pdispatch;
  98     struct pa_idxset *record_streams, *output_streams;
  99     uint32_t rrobin_index;
 100     struct pa_subscription *subscription;
 101 };
 102
 103 struct pa_protocol_native {
 104     struct pa_module *module;
 105     int public;
 106     struct pa_core *core;
 107     struct pa_socket_server *server;
 108     struct pa_idxset *connections;
 109     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
 110     int auth_cookie_in_property;
 111 };
 112
 113 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk);
 114 static void sink_input_drop_cb(struct pa_sink_input *i, const struct pa_memchunk *chunk, size_t length);
 115 static void sink_input_kill_cb(struct pa_sink_input *i);
 116 static pa_usec_t sink_input_get_latency_cb(struct pa_sink_input *i);
 117
 118 static void request_bytes(struct playback_stream*s);
 119
 120 static void source_output_kill_cb(struct pa_source_output *o);
 121 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk);
 122 static pa_usec_t source_output_get_latency_cb(struct pa_source_output *o);
 123
 124 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 125 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 126 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 127 static void command_create_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 128 static void command_delete_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 129 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 130 static void command_set_client_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 131 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 132 static void command_stat(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 133 static void command_get_playback_latency(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 134 static void command_get_record_latency(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 135 static void command_create_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 136 static void command_finish_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 137 static void command_play_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 138 static void command_remove_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 139 static void command_get_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 140 static void command_get_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 141 static void command_get_server_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 142 static void command_subscribe(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 143 static void command_set_volume(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 144 static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 145 static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 146 static void command_set_default_sink_or_source(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 147 static void command_set_stream_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 148 static void command_kill(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 149 static void command_load_module(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 150 static void command_unload_module(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 151 static void command_add_autoload(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 152 static void command_remove_autoload(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 153 static void command_get_autoload_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 154 static void command_get_autoload_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 155 static void command_cork_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 156 static void command_flush_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 157
 158 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
 159     [PA_COMMAND_ERROR] = { NULL },
 160     [PA_COMMAND_TIMEOUT] = { NULL },
 161     [PA_COMMAND_REPLY] = { NULL },
 162     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { command_create_playback_stream },
 163     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { command_delete_stream },
 164     [PA_COMMAND_DRAIN_PLAYBACK_STREAM] = { command_drain_playback_stream },
 165     [PA_COMMAND_CREATE_RECORD_STREAM] = { command_create_record_stream },
 166     [PA_COMMAND_DELETE_RECORD_STREAM] = { command_delete_stream },
 167     [PA_COMMAND_AUTH] = { command_auth },
 168     [PA_COMMAND_REQUEST] = { NULL },
 169     [PA_COMMAND_EXIT] = { command_exit },
 170     [PA_COMMAND_SET_CLIENT_NAME] = { command_set_client_name },
 171     [PA_COMMAND_LOOKUP_SINK] = { command_lookup },
 172     [PA_COMMAND_LOOKUP_SOURCE] = { command_lookup },
 173     [PA_COMMAND_STAT] = { command_stat },
 174     [PA_COMMAND_GET_PLAYBACK_LATENCY] = { command_get_playback_latency },
 175     [PA_COMMAND_GET_RECORD_LATENCY] = { command_get_record_latency },
 176     [PA_COMMAND_CREATE_UPLOAD_STREAM] = { command_create_upload_stream },
 177     [PA_COMMAND_DELETE_UPLOAD_STREAM] = { command_delete_stream },
 178     [PA_COMMAND_FINISH_UPLOAD_STREAM] = { command_finish_upload_stream },
 179     [PA_COMMAND_PLAY_SAMPLE] = { command_play_sample },
 180     [PA_COMMAND_REMOVE_SAMPLE] = { command_remove_sample },
 181     [PA_COMMAND_GET_SINK_INFO] = { command_get_info },
 182     [PA_COMMAND_GET_SOURCE_INFO] = { command_get_info },
 183     [PA_COMMAND_GET_CLIENT_INFO] = { command_get_info },
 184     [PA_COMMAND_GET_MODULE_INFO] = { command_get_info },
 185     [PA_COMMAND_GET_SINK_INPUT_INFO] = { command_get_info },
 186     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO] = { command_get_info },
 187     [PA_COMMAND_GET_SAMPLE_INFO] = { command_get_info },
 188     [PA_COMMAND_GET_SINK_INFO_LIST] = { command_get_info_list },
 189     [PA_COMMAND_GET_SOURCE_INFO_LIST] = { command_get_info_list },
 190     [PA_COMMAND_GET_MODULE_INFO_LIST] = { command_get_info_list },
 191     [PA_COMMAND_GET_CLIENT_INFO_LIST] = { command_get_info_list },
 192     [PA_COMMAND_GET_SINK_INPUT_INFO_LIST] = { command_get_info_list },
 193     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST] = { command_get_info_list },
 194     [PA_COMMAND_GET_SAMPLE_INFO_LIST] = { command_get_info_list },
 195     [PA_COMMAND_GET_SERVER_INFO] = { command_get_server_info },
 196     [PA_COMMAND_SUBSCRIBE] = { command_subscribe },
 197
 198     [PA_COMMAND_SET_SINK_VOLUME] = { command_set_volume },
 199     [PA_COMMAND_SET_SINK_INPUT_VOLUME] = { command_set_volume },
 200
 201     [PA_COMMAND_CORK_PLAYBACK_STREAM] = { command_cork_playback_stream },
 202     [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
 203     [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
 204     [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = { command_flush_or_trigger_playback_stream },
 205
 206     [PA_COMMAND_CORK_RECORD_STREAM] = { command_cork_record_stream },
 207     [PA_COMMAND_FLUSH_RECORD_STREAM] = { command_flush_record_stream },
 208
 209     [PA_COMMAND_SET_DEFAULT_SINK] = { command_set_default_sink_or_source },
 210     [PA_COMMAND_SET_DEFAULT_SOURCE] = { command_set_default_sink_or_source },
 211     [PA_COMMAND_SET_PLAYBACK_STREAM_NAME] = { command_set_stream_name },
 212     [PA_COMMAND_SET_RECORD_STREAM_NAME] = { command_set_stream_name },
 213     [PA_COMMAND_KILL_CLIENT] = { command_kill },
 214     [PA_COMMAND_KILL_SINK_INPUT] = { command_kill },
 215     [PA_COMMAND_KILL_SOURCE_OUTPUT] = { command_kill },
 216     [PA_COMMAND_LOAD_MODULE] = { command_load_module },
 217     [PA_COMMAND_UNLOAD_MODULE] = { command_unload_module },
 218     [PA_COMMAND_GET_AUTOLOAD_INFO] = { command_get_autoload_info },
 219     [PA_COMMAND_GET_AUTOLOAD_INFO_LIST] = { command_get_autoload_info_list },
 220     [PA_COMMAND_ADD_AUTOLOAD] = { command_add_autoload },
 221     [PA_COMMAND_REMOVE_AUTOLOAD] = { command_remove_autoload },
 222
 223 };
 224
 225 /* structure management */
 226
 227 static struct upload_stream* upload_stream_new(struct connection *c, const struct pa_sample_spec *ss, const char *name, size_t length) {
 228     struct upload_stream *s;
 229     assert(c && ss && name && length);
 230
 231     s = pa_xmalloc(sizeof(struct upload_stream));
 232     s->type = UPLOAD_STREAM;
 233     s->connection = c;
 234     s->sample_spec = *ss;
 235     s->name = pa_xstrdup(name);
 236
 237     s->memchunk.memblock = NULL;
 238     s->memchunk.index = 0;
 239     s->memchunk.length = 0;
 240
 241     s->length = length;
 242
 243     pa_idxset_put(c->output_streams, s, &s->index);
 244     return s;
 245 }
 246
 247 static void upload_stream_free(struct upload_stream *o) {
 248     assert(o && o->connection);
 249
 250     pa_idxset_remove_by_data(o->connection->output_streams, o, NULL);
 251
 252     pa_xfree(o->name);
 253
 254     if (o->memchunk.memblock)
 255         pa_memblock_unref(o->memchunk.memblock);
 256
 257     pa_xfree(o);
 258 }
 259
 260 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) {
 261     struct record_stream *s;
 262     struct pa_source_output *source_output;
 263     size_t base;
 264     assert(c && source && ss && name && maxlength);
 265
 266     if (!(source_output = pa_source_output_new(source, name, ss, -1)))
 267         return NULL;
 268
 269     s = pa_xmalloc(sizeof(struct record_stream));
 270     s->connection = c;
 271     s->source_output = source_output;
 272     s->source_output->push = source_output_push_cb;
 273     s->source_output->kill = source_output_kill_cb;
 274     s->source_output->get_latency = source_output_get_latency_cb;
 275     s->source_output->userdata = s;
 276     s->source_output->owner = c->protocol->module;
 277     s->source_output->client = c->client;
 278
 279     s->memblockq = pa_memblockq_new(maxlength, 0, base = pa_frame_size(ss), 0, 0, c->protocol->core->memblock_stat);
 280     assert(s->memblockq);
 281
 282     s->fragment_size = (fragment_size/base)*base;
 283     if (!s->fragment_size)
 284         s->fragment_size = base;
 285
 286     pa_idxset_put(c->record_streams, s, &s->index);
 287     return s;
 288 }
 289
 290 static void record_stream_free(struct record_stream* r) {
 291     assert(r && r->connection);
 292
 293     pa_idxset_remove_by_data(r->connection->record_streams, r, NULL);
 294     pa_source_output_disconnect(r->source_output);
 295     pa_source_output_unref(r->source_output);
 296     pa_memblockq_free(r->memblockq);
 297     pa_xfree(r);
 298 }
 299
 300 static struct playback_stream* playback_stream_new(struct connection *c, struct pa_sink *sink, const struct pa_sample_spec *ss, const char *name,
 301                                                    size_t maxlength,
 302                                                    size_t tlength,
 303                                                    size_t prebuf,
 304                                                    size_t minreq,
 305                                                    pa_volume_t volume) {
 306     struct playback_stream *s;
 307     struct pa_sink_input *sink_input;
 308     assert(c && sink && ss && name && maxlength);
 309
 310     if (!(sink_input = pa_sink_input_new(sink, name, ss, 0, -1)))
 311         return NULL;
 312
 313     s = pa_xmalloc(sizeof(struct playback_stream));
 314     s->type = PLAYBACK_STREAM;
 315     s->connection = c;
 316     s->sink_input = sink_input;
 317
 318     s->sink_input->peek = sink_input_peek_cb;
 319     s->sink_input->drop = sink_input_drop_cb;
 320     s->sink_input->kill = sink_input_kill_cb;
 321     s->sink_input->get_latency = sink_input_get_latency_cb;
 322     s->sink_input->userdata = s;
 323     s->sink_input->owner = c->protocol->module;
 324     s->sink_input->client = c->client;
 325
 326     s->memblockq = pa_memblockq_new(maxlength, tlength, pa_frame_size(ss), prebuf, minreq, c->protocol->core->memblock_stat);
 327     assert(s->memblockq);
 328
 329     s->requested_bytes = 0;
 330     s->drain_request = 0;
 331
 332     s->sink_input->volume = volume;
 333
 334     pa_idxset_put(c->output_streams, s, &s->index);
 335     return s;
 336 }
 337
 338 static void playback_stream_free(struct playback_stream* p) {
 339     assert(p && p->connection);
 340
 341     if (p->drain_request)
 342         pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERROR_NOENTITY);
 343
 344     pa_idxset_remove_by_data(p->connection->output_streams, p, NULL);
 345     pa_sink_input_disconnect(p->sink_input);
 346     pa_sink_input_unref(p->sink_input);
 347     pa_memblockq_free(p->memblockq);
 348     pa_xfree(p);
 349 }
 350
 351 static void connection_free(struct connection *c) {
 352     struct record_stream *r;
 353     struct output_stream *o;
 354     assert(c && c->protocol);
 355
 356     pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
 357     while ((r = pa_idxset_first(c->record_streams, NULL)))
 358         record_stream_free(r);
 359     pa_idxset_free(c->record_streams, NULL, NULL);
 360
 361     while ((o = pa_idxset_first(c->output_streams, NULL)))
 362         if (o->type == PLAYBACK_STREAM)
 363             playback_stream_free((struct playback_stream*) o);
 364         else
 365             upload_stream_free((struct upload_stream*) o);
 366     pa_idxset_free(c->output_streams, NULL, NULL);
 367
 368     pa_pdispatch_unref(c->pdispatch);
 369     pa_pstream_close(c->pstream);
 370     pa_pstream_unref(c->pstream);
 371     pa_client_free(c->client);
 372
 373     if (c->subscription)
 374         pa_subscription_free(c->subscription);
 375
 376     pa_xfree(c);
 377 }
 378
 379 static void request_bytes(struct playback_stream *s) {
 380     struct pa_tagstruct *t;
 381     size_t l;
 382     assert(s);
 383
 384     if (!(l = pa_memblockq_missing(s->memblockq)))
 385         return;
 386
 387     if (l <= s->requested_bytes)
 388         return;
 389
 390     l -= s->requested_bytes;
 391
 392     if (l < pa_memblockq_get_minreq(s->memblockq))
 393         return;
 394
 395     s->requested_bytes += l;
 396
 397     t = pa_tagstruct_new(NULL, 0);
 398     assert(t);
 399     pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 400     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 401     pa_tagstruct_putu32(t, s->index);
 402     pa_tagstruct_putu32(t, l);
 403     pa_pstream_send_tagstruct(s->connection->pstream, t);
 404
 405 /*     pa_log(__FILE__": Requesting %u bytes\n", l); */
 406 }
 407
 408 static void send_memblock(struct connection *c) {
 409     uint32_t start;
 410     struct record_stream *r;
 411
 412     start = PA_IDXSET_INVALID;
 413     for (;;) {
 414         struct pa_memchunk chunk;
 415
 416         if (!(r = pa_idxset_rrobin(c->record_streams, &c->rrobin_index)))
 417             return;
 418
 419         if (start == PA_IDXSET_INVALID)
 420             start = c->rrobin_index;
 421         else if (start == c->rrobin_index)
 422             return;
 423
 424         if (pa_memblockq_peek(r->memblockq,  &chunk) >= 0) {
 425             struct pa_memchunk schunk = chunk;
 426
 427             if (schunk.length > r->fragment_size)
 428                 schunk.length = r->fragment_size;
 429
 430             pa_pstream_send_memblock(c->pstream, r->index, 0, &schunk);
 431             pa_memblockq_drop(r->memblockq, &chunk, schunk.length);
 432             pa_memblock_unref(schunk.memblock);
 433
 434             return;
 435         }
 436     }
 437 }
 438
 439 static void send_playback_stream_killed(struct playback_stream *p) {
 440     struct pa_tagstruct *t;
 441     assert(p);
 442
 443     t = pa_tagstruct_new(NULL, 0);
 444     assert(t);
 445     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_KILLED);
 446     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 447     pa_tagstruct_putu32(t, p->index);
 448     pa_pstream_send_tagstruct(p->connection->pstream, t);
 449 }
 450
 451 static void send_record_stream_killed(struct record_stream *r) {
 452     struct pa_tagstruct *t;
 453     assert(r);
 454
 455     t = pa_tagstruct_new(NULL, 0);
 456     assert(t);
 457     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_KILLED);
 458     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 459     pa_tagstruct_putu32(t, r->index);
 460     pa_pstream_send_tagstruct(r->connection->pstream, t);
 461 }
 462
 463 /*** sinkinput callbacks ***/
 464
 465 static int sink_input_peek_cb(struct pa_sink_input *i, struct pa_memchunk *chunk) {
 466     struct playback_stream *s;
 467     assert(i && i->userdata && chunk);
 468     s = i->userdata;
 469
 470     if (pa_memblockq_peek(s->memblockq, chunk) < 0)
 471         return -1;
 472
 473     return 0;
 474 }
 475
 476 static void sink_input_drop_cb(struct pa_sink_input *i, const struct pa_memchunk *chunk, size_t length) {
 477     struct playback_stream *s;
 478     assert(i && i->userdata && length);
 479     s = i->userdata;
 480
 481     pa_memblockq_drop(s->memblockq, chunk, length);
 482     request_bytes(s);
 483
 484     if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) {
 485         pa_pstream_send_simple_ack(s->connection->pstream, s->drain_tag);
 486         s->drain_request = 0;
 487     }
 488
 489 /*     pa_log(__FILE__": after_drop: %u\n", pa_memblockq_get_length(s->memblockq)); */
 490 }
 491
 492 static void sink_input_kill_cb(struct pa_sink_input *i) {
 493     assert(i && i->userdata);
 494     send_playback_stream_killed((struct playback_stream *) i->userdata);
 495     playback_stream_free((struct playback_stream *) i->userdata);
 496 }
 497
 498 static pa_usec_t sink_input_get_latency_cb(struct pa_sink_input *i) {
 499     struct playback_stream *s;
 500     assert(i && i->userdata);
 501     s = i->userdata;
 502
 503     /*pa_log(__FILE__": get_latency: %u\n", pa_memblockq_get_length(s->memblockq));*/
 504
 505     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
 506 }
 507
 508 /*** source_output callbacks ***/
 509
 510 static void source_output_push_cb(struct pa_source_output *o, const struct pa_memchunk *chunk) {
 511     struct record_stream *s;
 512     assert(o && o->userdata && chunk);
 513     s = o->userdata;
 514
 515     pa_memblockq_push_align(s->memblockq, chunk, 0);
 516     if (!pa_pstream_is_pending(s->connection->pstream))
 517         send_memblock(s->connection);
 518 }
 519
 520 static void source_output_kill_cb(struct pa_source_output *o) {
 521     assert(o && o->userdata);
 522     send_record_stream_killed((struct record_stream *) o->userdata);
 523     record_stream_free((struct record_stream *) o->userdata);
 524 }
 525
 526 static pa_usec_t source_output_get_latency_cb(struct pa_source_output *o) {
 527     struct record_stream *s;
 528     assert(o && o->userdata);
 529     s = o->userdata;
 530
 531     /*pa_log(__FILE__": get_latency: %u\n", pa_memblockq_get_length(s->memblockq));*/
 532
 533     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &o->sample_spec);
 534 }
 535
 536 /*** pdispatch callbacks ***/
 537
 538 static void protocol_error(struct connection *c) {
 539     pa_log(__FILE__": protocol error, kicking client\n");
 540     connection_free(c);
 541 }
 542
 543 static void command_create_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 544     struct connection *c = userdata;
 545     struct playback_stream *s;
 546     size_t maxlength, tlength, prebuf, minreq;
 547     uint32_t sink_index;
 548     const char *name, *sink_name;
 549     struct pa_sample_spec ss;
 550     struct pa_tagstruct *reply;
 551     struct pa_sink *sink;
 552     pa_volume_t volume;
 553     int corked;
 554     assert(c && t && c->protocol && c->protocol->core);
 555
 556     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
 557         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 558         pa_tagstruct_getu32(t, &sink_index) < 0 ||
 559         pa_tagstruct_gets(t, &sink_name) < 0 ||
 560         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 561         pa_tagstruct_get_boolean(t, &corked) < 0 ||
 562         pa_tagstruct_getu32(t, &tlength) < 0 ||
 563         pa_tagstruct_getu32(t, &prebuf) < 0 ||
 564         pa_tagstruct_getu32(t, &minreq) < 0 ||
 565         pa_tagstruct_getu32(t, &volume) < 0 ||
 566         !pa_tagstruct_eof(t)) {
 567         protocol_error(c);
 568         return;
 569     }
 570
 571
 572     if (!c->authorized) {
 573         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 574         return;
 575     }
 576
 577     if (sink_index != (uint32_t) -1)
 578         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
 579     else
 580         sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1);
 581
 582     if (!sink) {
 583         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 584         return;
 585     }
 586
 587     if (!(s = playback_stream_new(c, sink, &ss, name, maxlength, tlength, prebuf, minreq, volume))) {
 588         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 589         return;
 590     }
 591
 592     pa_sink_input_cork(s->sink_input, corked);
 593
 594     reply = pa_tagstruct_new(NULL, 0);
 595     assert(reply);
 596     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 597     pa_tagstruct_putu32(reply, tag);
 598     pa_tagstruct_putu32(reply, s->index);
 599     assert(s->sink_input);
 600     pa_tagstruct_putu32(reply, s->sink_input->index);
 601     pa_tagstruct_putu32(reply, s->requested_bytes = pa_memblockq_missing(s->memblockq));
 602     pa_pstream_send_tagstruct(c->pstream, reply);
 603     request_bytes(s);
 604 }
 605
 606 static void command_delete_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 607     struct connection *c = userdata;
 608     uint32_t channel;
 609     assert(c && t);
 610
 611     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 612         !pa_tagstruct_eof(t)) {
 613         protocol_error(c);
 614         return;
 615     }
 616
 617     if (!c->authorized) {
 618         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 619         return;
 620     }
 621
 622     if (command == PA_COMMAND_DELETE_PLAYBACK_STREAM) {
 623         struct playback_stream *s;
 624         if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != PLAYBACK_STREAM)) {
 625             pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 626             return;
 627         }
 628
 629         playback_stream_free(s);
 630     } else if (command == PA_COMMAND_DELETE_RECORD_STREAM) {
 631         struct record_stream *s;
 632         if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
 633             pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 634             return;
 635         }
 636
 637         record_stream_free(s);
 638     } else {
 639         struct upload_stream *s;
 640         assert(command == PA_COMMAND_DELETE_UPLOAD_STREAM);
 641         if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) {
 642             pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
 643             return;
 644         }
 645
 646         upload_stream_free(s);
 647     }
 648
 649     pa_pstream_send_simple_ack(c->pstream, tag);
 650 }
 651
 652 static void command_create_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 653     struct connection *c = userdata;
 654     struct record_stream *s;
 655     size_t maxlength, fragment_size;
 656     uint32_t source_index;
 657     const char *name, *source_name;
 658     struct pa_sample_spec ss;
 659     struct pa_tagstruct *reply;
 660     struct pa_source *source;
 661     int corked;
 662     assert(c && t && c->protocol && c->protocol->core);
 663
 664     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
 665         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 666         pa_tagstruct_getu32(t, &source_index) < 0 ||
 667         pa_tagstruct_gets(t, &source_name) < 0 ||
 668         pa_tagstruct_getu32(t, &maxlength) < 0 ||
 669         pa_tagstruct_get_boolean(t, &corked) < 0 ||
 670         pa_tagstruct_getu32(t, &fragment_size) < 0 ||
 671         !pa_tagstruct_eof(t)) {
 672         protocol_error(c);
 673         return;
 674     }
 675
 676     if (!c->authorized) {
 677         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 678         return;
 679     }
 680
 681     if (source_index != (uint32_t) -1)
 682         source = pa_idxset_get_by_index(c->protocol->core->sources, source_index);
 683     else
 684         source = pa_namereg_get(c->protocol->core, source_name, PA_NAMEREG_SOURCE, 1);
 685
 686     if (!source) {
 687         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 688         return;
 689     }
 690
 691     if (!(s = record_stream_new(c, source, &ss, name, maxlength, fragment_size))) {
 692         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 693         return;
 694     }
 695
 696     pa_source_output_cork(s->source_output, corked);
 697
 698     reply = pa_tagstruct_new(NULL, 0);
 699     assert(reply);
 700     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 701     pa_tagstruct_putu32(reply, tag);
 702     pa_tagstruct_putu32(reply, s->index);
 703     assert(s->source_output);
 704     pa_tagstruct_putu32(reply, s->source_output->index);
 705     pa_pstream_send_tagstruct(c->pstream, reply);
 706 }
 707
 708 static void command_exit(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 709     struct connection *c = userdata;
 710     assert(c && t);
 711
 712     if (!pa_tagstruct_eof(t)) {
 713         protocol_error(c);
 714         return;
 715     }
 716
 717     if (!c->authorized) {
 718         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 719         return;
 720     }
 721
 722     assert(c->protocol && c->protocol->core && c->protocol->core->mainloop);
 723     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
 724     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
 725     return;
 726 }
 727
 728 static void command_auth(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 729     struct connection *c = userdata;
 730     const void*cookie;
 731     assert(c && t);
 732
 733     if (pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
 734         !pa_tagstruct_eof(t)) {
 735         protocol_error(c);
 736         return;
 737     }
 738
 739     if (!c->authorized) {
 740         if (memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) != 0) {
 741             pa_log(__FILE__": Denied access to client with invalid authorization key.\n");
 742             pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 743             return;
 744         }
 745
 746         c->authorized = 1;
 747     }
 748
 749     pa_pstream_send_simple_ack(c->pstream, tag);
 750     return;
 751 }
 752
 753 static void command_set_client_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 754     struct connection *c = userdata;
 755     const char *name;
 756     assert(c && t);
 757
 758     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
 759         !pa_tagstruct_eof(t)) {
 760         protocol_error(c);
 761         return;
 762     }
 763
 764     pa_client_set_name(c->client, name);
 765     pa_pstream_send_simple_ack(c->pstream, tag);
 766     return;
 767 }
 768
 769 static void command_lookup(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 770     struct connection *c = userdata;
 771     const char *name;
 772     uint32_t index = PA_IDXSET_INVALID;
 773     assert(c && t);
 774
 775     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
 776         !pa_tagstruct_eof(t)) {
 777         protocol_error(c);
 778         return;
 779     }
 780
 781     if (!c->authorized) {
 782         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 783         return;
 784     }
 785
 786     if (command == PA_COMMAND_LOOKUP_SINK) {
 787         struct pa_sink *sink;
 788         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1)))
 789             index = sink->index;
 790     } else {
 791         struct pa_source *source;
 792         assert(command == PA_COMMAND_LOOKUP_SOURCE);
 793         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1)))
 794             index = source->index;
 795     }
 796
 797     if (index == PA_IDXSET_INVALID)
 798         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 799     else {
 800         struct pa_tagstruct *reply;
 801         reply = pa_tagstruct_new(NULL, 0);
 802         assert(reply);
 803         pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 804         pa_tagstruct_putu32(reply, tag);
 805         pa_tagstruct_putu32(reply, index);
 806         pa_pstream_send_tagstruct(c->pstream, reply);
 807     }
 808 }
 809
 810 static void command_drain_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 811     struct connection *c = userdata;
 812     uint32_t index;
 813     struct playback_stream *s;
 814     assert(c && t);
 815
 816     if (pa_tagstruct_getu32(t, &index) < 0 ||
 817         !pa_tagstruct_eof(t)) {
 818         protocol_error(c);
 819         return;
 820     }
 821
 822     if (!c->authorized) {
 823         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 824         return;
 825     }
 826
 827     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
 828         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 829         return;
 830     }
 831
 832     s->drain_request = 0;
 833
 834     pa_memblockq_prebuf_disable(s->memblockq);
 835
 836     if (!pa_memblockq_is_readable(s->memblockq)) {
 837 /*         pa_log("immediate drain: %u\n", pa_memblockq_get_length(s->memblockq)); */
 838         pa_pstream_send_simple_ack(c->pstream, tag);
 839     } else {
 840 /*         pa_log("slow drain triggered\n"); */
 841         s->drain_request = 1;
 842         s->drain_tag = tag;
 843
 844         pa_sink_notify(s->sink_input->sink);
 845     }
 846 }
 847
 848 static void command_stat(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 849     struct connection *c = userdata;
 850     struct pa_tagstruct *reply;
 851     assert(c && t);
 852
 853     if (!pa_tagstruct_eof(t)) {
 854         protocol_error(c);
 855         return;
 856     }
 857
 858     if (!c->authorized) {
 859         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 860         return;
 861     }
 862
 863     reply = pa_tagstruct_new(NULL, 0);
 864     assert(reply);
 865     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 866     pa_tagstruct_putu32(reply, tag);
 867     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->total);
 868     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->total_size);
 869     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->allocated);
 870     pa_tagstruct_putu32(reply, c->protocol->core->memblock_stat->allocated_size);
 871     pa_tagstruct_putu32(reply, pa_scache_total_size(c->protocol->core));
 872     pa_pstream_send_tagstruct(c->pstream, reply);
 873 }
 874
 875 static void command_get_playback_latency(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 876     struct connection *c = userdata;
 877     struct pa_tagstruct *reply;
 878     struct playback_stream *s;
 879     struct timeval tv, now;
 880     uint64_t counter;
 881     uint32_t index;
 882     assert(c && t);
 883
 884     if (pa_tagstruct_getu32(t, &index) < 0 ||
 885         pa_tagstruct_get_timeval(t, &tv) < 0 ||
 886         pa_tagstruct_getu64(t, &counter) < 0 ||
 887         !pa_tagstruct_eof(t)) {
 888         protocol_error(c);
 889         return;
 890     }
 891
 892     if (!c->authorized) {
 893         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 894         return;
 895     }
 896
 897     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
 898         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 899         return;
 900     }
 901
 902     reply = pa_tagstruct_new(NULL, 0);
 903     assert(reply);
 904     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 905     pa_tagstruct_putu32(reply, tag);
 906     pa_tagstruct_put_usec(reply, pa_sink_input_get_latency(s->sink_input));
 907     pa_tagstruct_put_usec(reply, pa_sink_get_latency(s->sink_input->sink));
 908     pa_tagstruct_put_usec(reply, 0);
 909     pa_tagstruct_put_boolean(reply, pa_memblockq_is_readable(s->memblockq));
 910     pa_tagstruct_putu32(reply, pa_memblockq_get_length(s->memblockq));
 911     pa_tagstruct_put_timeval(reply, &tv);
 912     gettimeofday(&now, NULL);
 913     pa_tagstruct_put_timeval(reply, &now);
 914     pa_tagstruct_putu64(reply, counter);
 915     pa_pstream_send_tagstruct(c->pstream, reply);
 916 }
 917
 918 static void command_get_record_latency(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 919     struct connection *c = userdata;
 920     struct pa_tagstruct *reply;
 921     struct record_stream *s;
 922     struct timeval tv, now;
 923     uint64_t counter;
 924     uint32_t index;
 925     assert(c && t);
 926
 927     if (pa_tagstruct_getu32(t, &index) < 0 ||
 928         pa_tagstruct_get_timeval(t, &tv) < 0 ||
 929         pa_tagstruct_getu64(t, &counter) < 0 ||
 930         !pa_tagstruct_eof(t)) {
 931         protocol_error(c);
 932         return;
 933     }
 934
 935     if (!c->authorized) {
 936         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 937         return;
 938     }
 939
 940     if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
 941         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
 942         return;
 943     }
 944
 945     reply = pa_tagstruct_new(NULL, 0);
 946     assert(reply);
 947     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 948     pa_tagstruct_putu32(reply, tag);
 949     pa_tagstruct_put_usec(reply, pa_source_output_get_latency(s->source_output));
 950     pa_tagstruct_put_usec(reply, s->source_output->source->monitor_of ? pa_sink_get_latency(s->source_output->source->monitor_of) : 0);
 951     pa_tagstruct_put_usec(reply, pa_source_get_latency(s->source_output->source));
 952     pa_tagstruct_put_boolean(reply, 0);
 953     pa_tagstruct_putu32(reply, pa_memblockq_get_length(s->memblockq));
 954     pa_tagstruct_put_timeval(reply, &tv);
 955     gettimeofday(&now, NULL);
 956     pa_tagstruct_put_timeval(reply, &now);
 957     pa_tagstruct_putu64(reply, counter);
 958     pa_pstream_send_tagstruct(c->pstream, reply);
 959 }
 960
 961
 962 static void command_create_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 963     struct connection *c = userdata;
 964     struct upload_stream *s;
 965     size_t length;
 966     const char *name;
 967     struct pa_sample_spec ss;
 968     struct pa_tagstruct *reply;
 969     assert(c && t && c->protocol && c->protocol->core);
 970
 971     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
 972         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
 973         pa_tagstruct_getu32(t, &length) < 0 ||
 974         !pa_tagstruct_eof(t)) {
 975         protocol_error(c);
 976         return;
 977     }
 978
 979     if (!c->authorized) {
 980         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
 981         return;
 982     }
 983
 984     if ((length % pa_frame_size(&ss)) != 0 || length <= 0 || !*name) {
 985         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 986         return;
 987     }
 988
 989     if (!(s = upload_stream_new(c, &ss, name, length))) {
 990         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INVALID);
 991         return;
 992     }
 993
 994     reply = pa_tagstruct_new(NULL, 0);
 995     assert(reply);
 996     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
 997     pa_tagstruct_putu32(reply, tag);
 998     pa_tagstruct_putu32(reply, s->index);
 999     pa_tagstruct_putu32(reply, length);
1000     pa_pstream_send_tagstruct(c->pstream, reply);
1001 }
1002
1003 static void command_finish_upload_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1004     struct connection *c = userdata;
1005     uint32_t channel;
1006     struct upload_stream *s;
1007     uint32_t index;
1008     assert(c && t);
1009
1010     if (pa_tagstruct_getu32(t, &channel) < 0 ||
1011         !pa_tagstruct_eof(t)) {
1012         protocol_error(c);
1013         return;
1014     }
1015
1016     if (!c->authorized) {
1017         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1018         return;
1019     }
1020
1021     if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) {
1022         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
1023         return;
1024     }
1025
1026     pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->memchunk, &index);
1027     pa_pstream_send_simple_ack(c->pstream, tag);
1028     upload_stream_free(s);
1029 }
1030
1031 static void command_play_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1032     struct connection *c = userdata;
1033     uint32_t sink_index, volume;
1034     struct pa_sink *sink;
1035     const char *name, *sink_name;
1036     assert(c && t);
1037
1038     if (pa_tagstruct_getu32(t, &sink_index) < 0 ||
1039         pa_tagstruct_gets(t, &sink_name) < 0 ||
1040         pa_tagstruct_getu32(t, &volume) < 0 ||
1041         pa_tagstruct_gets(t, &name) < 0 || !name ||
1042         !pa_tagstruct_eof(t)) {
1043         protocol_error(c);
1044         return;
1045     }
1046
1047     if (!c->authorized) {
1048         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1049         return;
1050     }
1051
1052     if (sink_index != (uint32_t) -1)
1053         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
1054     else
1055         sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1);
1056
1057     if (!sink) {
1058         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1059         return;
1060     }
1061
1062     if (pa_scache_play_item(c->protocol->core, name, sink, volume) < 0) {
1063         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1064         return;
1065     }
1066
1067     pa_pstream_send_simple_ack(c->pstream, tag);
1068 }
1069
1070 static void command_remove_sample(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1071     struct connection *c = userdata;
1072     const char *name;
1073     assert(c && t);
1074
1075     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
1076         !pa_tagstruct_eof(t)) {
1077         protocol_error(c);
1078         return;
1079     }
1080
1081     if (!c->authorized) {
1082         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1083         return;
1084     }
1085
1086     if (pa_scache_remove_item(c->protocol->core, name) < 0) {
1087         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1088         return;
1089     }
1090
1091     pa_pstream_send_simple_ack(c->pstream, tag);
1092 }
1093
1094 static void sink_fill_tagstruct(struct pa_tagstruct *t, struct pa_sink *sink) {
1095     assert(t && sink);
1096     pa_tagstruct_putu32(t, sink->index);
1097     pa_tagstruct_puts(t, sink->name);
1098     pa_tagstruct_puts(t, sink->description);
1099     pa_tagstruct_put_sample_spec(t, &sink->sample_spec);
1100     pa_tagstruct_putu32(t, sink->owner ? sink->owner->index : (uint32_t) -1);
1101     pa_tagstruct_putu32(t, sink->volume);
1102     pa_tagstruct_putu32(t, sink->monitor_source->index);
1103     pa_tagstruct_puts(t, sink->monitor_source->name);
1104     pa_tagstruct_put_usec(t, pa_sink_get_latency(sink));
1105 }
1106
1107 static void source_fill_tagstruct(struct pa_tagstruct *t, struct pa_source *source) {
1108     assert(t && source);
1109     pa_tagstruct_putu32(t, source->index);
1110     pa_tagstruct_puts(t, source->name);
1111     pa_tagstruct_puts(t, source->description);
1112     pa_tagstruct_put_sample_spec(t, &source->sample_spec);
1113     pa_tagstruct_putu32(t, source->owner ? source->owner->index : (uint32_t) -1);
1114     pa_tagstruct_putu32(t, source->monitor_of ? source->monitor_of->index : (uint32_t) -1);
1115     pa_tagstruct_puts(t, source->monitor_of ? source->monitor_of->name : NULL);
1116     pa_tagstruct_put_usec(t, pa_source_get_latency(source));
1117 }
1118
1119 static void client_fill_tagstruct(struct pa_tagstruct *t, struct pa_client *client) {
1120     assert(t && client);
1121     pa_tagstruct_putu32(t, client->index);
1122     pa_tagstruct_puts(t, client->name);
1123     pa_tagstruct_puts(t, client->protocol_name);
1124     pa_tagstruct_putu32(t, client->owner ? client->owner->index : (uint32_t) -1);
1125 }
1126
1127 static void module_fill_tagstruct(struct pa_tagstruct *t, struct pa_module *module) {
1128     assert(t && module);
1129     pa_tagstruct_putu32(t, module->index);
1130     pa_tagstruct_puts(t, module->name);
1131     pa_tagstruct_puts(t, module->argument);
1132     pa_tagstruct_putu32(t, module->n_used);
1133     pa_tagstruct_put_boolean(t, module->auto_unload);
1134 }
1135
1136 static void sink_input_fill_tagstruct(struct pa_tagstruct *t, struct pa_sink_input *s) {
1137     assert(t && s);
1138     pa_tagstruct_putu32(t, s->index);
1139     pa_tagstruct_puts(t, s->name);
1140     pa_tagstruct_putu32(t, s->owner ? s->owner->index : (uint32_t) -1);
1141     pa_tagstruct_putu32(t, s->client ? s->client->index : (uint32_t) -1);
1142     pa_tagstruct_putu32(t, s->sink->index);
1143     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1144     pa_tagstruct_putu32(t, s->volume);
1145     pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s));
1146     pa_tagstruct_put_usec(t, pa_sink_get_latency(s->sink));
1147 }
1148
1149 static void source_output_fill_tagstruct(struct pa_tagstruct *t, struct pa_source_output *s) {
1150     assert(t && s);
1151     pa_tagstruct_putu32(t, s->index);
1152     pa_tagstruct_puts(t, s->name);
1153     pa_tagstruct_putu32(t, s->owner ? s->owner->index : (uint32_t) -1);
1154     pa_tagstruct_putu32(t, s->client ? s->client->index : (uint32_t) -1);
1155     pa_tagstruct_putu32(t, s->source->index);
1156     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1157     pa_tagstruct_put_usec(t, pa_source_output_get_latency(s));
1158     pa_tagstruct_put_usec(t, pa_source_get_latency(s->source));
1159 }
1160
1161 static void scache_fill_tagstruct(struct pa_tagstruct *t, struct pa_scache_entry *e) {
1162     assert(t && e);
1163     pa_tagstruct_putu32(t, e->index);
1164     pa_tagstruct_puts(t, e->name);
1165     pa_tagstruct_putu32(t, e->volume);
1166     pa_tagstruct_put_usec(t, pa_bytes_to_usec(e->memchunk.length, &e->sample_spec));
1167     pa_tagstruct_put_sample_spec(t, &e->sample_spec);
1168     pa_tagstruct_putu32(t, e->memchunk.length);
1169     pa_tagstruct_put_boolean(t, e->lazy);
1170     pa_tagstruct_puts(t, e->filename);
1171 }
1172
1173 static void command_get_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1174     struct connection *c = userdata;
1175     uint32_t index;
1176     struct pa_sink *sink = NULL;
1177     struct pa_source *source = NULL;
1178     struct pa_client *client = NULL;
1179     struct pa_module *module = NULL;
1180     struct pa_sink_input *si = NULL;
1181     struct pa_source_output *so = NULL;
1182     struct pa_scache_entry *sce = NULL;
1183     const char *name;
1184     struct pa_tagstruct *reply;
1185     assert(c && t);
1186
1187
1188     if (pa_tagstruct_getu32(t, &index) < 0 ||
1189         (command != PA_COMMAND_GET_CLIENT_INFO &&
1190          command != PA_COMMAND_GET_MODULE_INFO &&
1191          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
1192          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
1193          pa_tagstruct_gets(t, &name) < 0) ||
1194         !pa_tagstruct_eof(t)) {
1195         protocol_error(c);
1196         return;
1197     }
1198
1199     if (!c->authorized) {
1200         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1201         return;
1202     }
1203
1204     if (command == PA_COMMAND_GET_SINK_INFO) {
1205         if (index != (uint32_t) -1)
1206             sink = pa_idxset_get_by_index(c->protocol->core->sinks, index);
1207         else
1208             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
1209     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
1210         if (index != (uint32_t) -1)
1211             source = pa_idxset_get_by_index(c->protocol->core->sources, index);
1212         else
1213             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
1214     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
1215         client = pa_idxset_get_by_index(c->protocol->core->clients, index);
1216     else if (command == PA_COMMAND_GET_MODULE_INFO)
1217         module = pa_idxset_get_by_index(c->protocol->core->modules, index);
1218     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
1219         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index);
1220     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
1221         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, index);
1222     else {
1223         assert(command == PA_COMMAND_GET_SAMPLE_INFO);
1224         if (index != (uint32_t) -1)
1225             sce = pa_idxset_get_by_index(c->protocol->core->scache, index);
1226         else
1227             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE, 0);
1228     }
1229
1230     if (!sink && !source && !client && !module && !si && !so && !sce) {
1231         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1232         return;
1233     }
1234
1235     reply = pa_tagstruct_new(NULL, 0);
1236     assert(reply);
1237     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1238     pa_tagstruct_putu32(reply, tag);
1239     if (sink)
1240         sink_fill_tagstruct(reply, sink);
1241     else if (source)
1242         source_fill_tagstruct(reply, source);
1243     else if (client)
1244         client_fill_tagstruct(reply, client);
1245     else if (module)
1246         module_fill_tagstruct(reply, module);
1247     else if (si)
1248         sink_input_fill_tagstruct(reply, si);
1249     else if (so)
1250         source_output_fill_tagstruct(reply, so);
1251     else
1252         scache_fill_tagstruct(reply, sce);
1253     pa_pstream_send_tagstruct(c->pstream, reply);
1254 }
1255
1256 static void command_get_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1257     struct connection *c = userdata;
1258     struct pa_idxset *i;
1259     uint32_t index;
1260     void *p;
1261     struct pa_tagstruct *reply;
1262     assert(c && t);
1263
1264     if (!pa_tagstruct_eof(t)) {
1265         protocol_error(c);
1266         return;
1267     }
1268
1269     if (!c->authorized) {
1270         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1271         return;
1272     }
1273
1274     reply = pa_tagstruct_new(NULL, 0);
1275     assert(reply);
1276     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1277     pa_tagstruct_putu32(reply, tag);
1278
1279     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
1280         i = c->protocol->core->sinks;
1281     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
1282         i = c->protocol->core->sources;
1283     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
1284         i = c->protocol->core->clients;
1285     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
1286         i = c->protocol->core->modules;
1287     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
1288         i = c->protocol->core->sink_inputs;
1289     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
1290         i = c->protocol->core->source_outputs;
1291     else {
1292         assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
1293         i = c->protocol->core->scache;
1294     }
1295
1296     if (i) {
1297         for (p = pa_idxset_first(i, &index); p; p = pa_idxset_next(i, &index)) {
1298             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
1299                 sink_fill_tagstruct(reply, p);
1300             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
1301                 source_fill_tagstruct(reply, p);
1302             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
1303                 client_fill_tagstruct(reply, p);
1304             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
1305                 module_fill_tagstruct(reply, p);
1306             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
1307                 sink_input_fill_tagstruct(reply, p);
1308             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
1309                 source_output_fill_tagstruct(reply, p);
1310             else {
1311                 assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
1312                 scache_fill_tagstruct(reply, p);
1313             }
1314         }
1315     }
1316
1317     pa_pstream_send_tagstruct(c->pstream, reply);
1318 }
1319
1320 static void command_get_server_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1321     struct connection *c = userdata;
1322     struct pa_tagstruct *reply;
1323     char txt[256];
1324     const char *n;
1325     assert(c && t);
1326
1327     if (!pa_tagstruct_eof(t)) {
1328         protocol_error(c);
1329         return;
1330     }
1331
1332     if (!c->authorized) {
1333         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1334         return;
1335     }
1336
1337     reply = pa_tagstruct_new(NULL, 0);
1338     assert(reply);
1339     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1340     pa_tagstruct_putu32(reply, tag);
1341     pa_tagstruct_puts(reply, PACKAGE_NAME);
1342     pa_tagstruct_puts(reply, PACKAGE_VERSION);
1343     pa_tagstruct_puts(reply, pa_get_user_name(txt, sizeof(txt)));
1344     pa_tagstruct_puts(reply, pa_get_host_name(txt, sizeof(txt)));
1345     pa_tagstruct_put_sample_spec(reply, &c->protocol->core->default_sample_spec);
1346
1347     n = pa_namereg_get_default_sink_name(c->protocol->core);
1348     pa_tagstruct_puts(reply, n);
1349     n = pa_namereg_get_default_source_name(c->protocol->core);
1350     pa_tagstruct_puts(reply, n);
1351     pa_pstream_send_tagstruct(c->pstream, reply);
1352 }
1353
1354 static void subscription_cb(struct pa_core *core, enum pa_subscription_event_type e, uint32_t index, void *userdata) {
1355     struct pa_tagstruct *t;
1356     struct connection *c = userdata;
1357     assert(c && core);
1358
1359     t = pa_tagstruct_new(NULL, 0);
1360     assert(t);
1361     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
1362     pa_tagstruct_putu32(t, (uint32_t) -1);
1363     pa_tagstruct_putu32(t, e);
1364     pa_tagstruct_putu32(t, index);
1365     pa_pstream_send_tagstruct(c->pstream, t);
1366 }
1367
1368 static void command_subscribe(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1369     struct connection *c = userdata;
1370     enum pa_subscription_mask m;
1371     assert(c && t);
1372
1373     if (pa_tagstruct_getu32(t, &m) < 0 ||
1374         !pa_tagstruct_eof(t)) {
1375         protocol_error(c);
1376         return;
1377     }
1378
1379     if (!c->authorized) {
1380         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1381         return;
1382     }
1383
1384     if (c->subscription)
1385         pa_subscription_free(c->subscription);
1386
1387     if (m != 0) {
1388         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
1389         assert(c->subscription);
1390     } else
1391         c->subscription = NULL;
1392
1393     pa_pstream_send_simple_ack(c->pstream, tag);
1394 }
1395
1396 static void command_set_volume(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1397     struct connection *c = userdata;
1398     uint32_t index, volume;
1399     struct pa_sink *sink = NULL;
1400     struct pa_sink_input *si = NULL;
1401     const char *name = NULL;
1402     assert(c && t);
1403
1404     if (pa_tagstruct_getu32(t, &index) < 0 ||
1405         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
1406         pa_tagstruct_getu32(t, &volume) ||
1407         !pa_tagstruct_eof(t)) {
1408         protocol_error(c);
1409         return;
1410     }
1411
1412     if (!c->authorized) {
1413         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1414         return;
1415     }
1416
1417     if (command == PA_COMMAND_SET_SINK_VOLUME) {
1418         if (index != (uint32_t) -1)
1419             sink = pa_idxset_get_by_index(c->protocol->core->sinks, index);
1420         else
1421             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
1422     }  else {
1423         assert(command == PA_COMMAND_SET_SINK_INPUT_VOLUME);
1424         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index);
1425     }
1426
1427     if (!si && !sink) {
1428         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1429         return;
1430     }
1431
1432     if (sink)
1433         pa_sink_set_volume(sink, volume);
1434     else if (si)
1435         pa_sink_input_set_volume(si, volume);
1436
1437     pa_pstream_send_simple_ack(c->pstream, tag);
1438 }
1439
1440 static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1441     struct connection *c = userdata;
1442     uint32_t index;
1443     int b;
1444     struct playback_stream *s;
1445     assert(c && t);
1446
1447     if (pa_tagstruct_getu32(t, &index) < 0 ||
1448         pa_tagstruct_get_boolean(t, &b) < 0 ||
1449         !pa_tagstruct_eof(t)) {
1450         protocol_error(c);
1451         return;
1452     }
1453
1454     if (!c->authorized) {
1455         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1456         return;
1457     }
1458
1459     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1460         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1461         return;
1462     }
1463
1464     pa_sink_input_cork(s->sink_input, b);
1465     pa_pstream_send_simple_ack(c->pstream, tag);
1466 }
1467
1468 static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1469     struct connection *c = userdata;
1470     uint32_t index;
1471     struct playback_stream *s;
1472     assert(c && t);
1473
1474     if (pa_tagstruct_getu32(t, &index) < 0 ||
1475         !pa_tagstruct_eof(t)) {
1476         protocol_error(c);
1477         return;
1478     }
1479
1480     if (!c->authorized) {
1481         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1482         return;
1483     }
1484
1485     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1486         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1487         return;
1488     }
1489
1490     if (command == PA_COMMAND_PREBUF_PLAYBACK_STREAM)
1491         pa_memblockq_prebuf_reenable(s->memblockq);
1492     else if (command == PA_COMMAND_TRIGGER_PLAYBACK_STREAM)
1493         pa_memblockq_prebuf_disable(s->memblockq);
1494     else {
1495         assert(command == PA_COMMAND_FLUSH_PLAYBACK_STREAM);
1496         pa_memblockq_flush(s->memblockq);
1497         /*pa_log(__FILE__": flush: %u\n", pa_memblockq_get_length(s->memblockq));*/
1498     }
1499
1500     pa_sink_notify(s->sink_input->sink);
1501     pa_pstream_send_simple_ack(c->pstream, tag);
1502     request_bytes(s);
1503 }
1504
1505 static void command_cork_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1506     struct connection *c = userdata;
1507     uint32_t index;
1508     struct record_stream *s;
1509     int b;
1510     assert(c && t);
1511
1512     if (pa_tagstruct_getu32(t, &index) < 0 ||
1513         pa_tagstruct_get_boolean(t, &b) < 0 ||
1514         !pa_tagstruct_eof(t)) {
1515         protocol_error(c);
1516         return;
1517     }
1518
1519     if (!c->authorized) {
1520         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1521         return;
1522     }
1523
1524     if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
1525         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1526         return;
1527     }
1528
1529     pa_source_output_cork(s->source_output, b);
1530     pa_pstream_send_simple_ack(c->pstream, tag);
1531 }
1532
1533 static void command_flush_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1534     struct connection *c = userdata;
1535     uint32_t index;
1536     struct record_stream *s;
1537     assert(c && t);
1538
1539     if (pa_tagstruct_getu32(t, &index) < 0 ||
1540         !pa_tagstruct_eof(t)) {
1541         protocol_error(c);
1542         return;
1543     }
1544
1545     if (!c->authorized) {
1546         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1547         return;
1548     }
1549
1550     if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
1551         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1552         return;
1553     }
1554
1555     pa_memblockq_flush(s->memblockq);
1556     pa_pstream_send_simple_ack(c->pstream, tag);
1557 }
1558
1559 static void command_set_default_sink_or_source(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1560     struct connection *c = userdata;
1561     uint32_t index;
1562     const char *s;
1563     assert(c && t);
1564
1565     if (pa_tagstruct_getu32(t, &index) < 0 ||
1566         pa_tagstruct_gets(t, &s) < 0 || !s ||
1567         !pa_tagstruct_eof(t)) {
1568         protocol_error(c);
1569         return;
1570     }
1571
1572     if (!c->authorized) {
1573         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1574         return;
1575     }
1576
1577     pa_namereg_set_default(c->protocol->core, s, command == PA_COMMAND_SET_DEFAULT_SOURCE ? PA_NAMEREG_SOURCE : PA_NAMEREG_SINK);
1578     pa_pstream_send_simple_ack(c->pstream, tag);
1579 }
1580
1581 static void command_set_stream_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1582     struct connection *c = userdata;
1583     uint32_t index;
1584     const char *name;
1585     assert(c && t);
1586
1587     if (pa_tagstruct_getu32(t, &index) < 0 ||
1588         pa_tagstruct_gets(t, &name) < 0 || !name ||
1589         !pa_tagstruct_eof(t)) {
1590         protocol_error(c);
1591         return;
1592     }
1593
1594     if (!c->authorized) {
1595         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1596         return;
1597     }
1598
1599     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
1600         struct playback_stream *s;
1601
1602         if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1603             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1604             return;
1605         }
1606
1607         pa_sink_input_set_name(s->sink_input, name);
1608
1609     } else {
1610         struct record_stream *s;
1611
1612         if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
1613             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1614             return;
1615         }
1616
1617         pa_source_output_set_name(s->source_output, name);
1618     }
1619
1620     pa_pstream_send_simple_ack(c->pstream, tag);
1621 }
1622
1623 static void command_kill(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1624     struct connection *c = userdata;
1625     uint32_t index;
1626     assert(c && t);
1627
1628     if (pa_tagstruct_getu32(t, &index) < 0 ||
1629         !pa_tagstruct_eof(t)) {
1630         protocol_error(c);
1631         return;
1632     }
1633
1634     if (!c->authorized) {
1635         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1636         return;
1637     }
1638
1639     if (command == PA_COMMAND_KILL_CLIENT) {
1640         struct pa_client *client;
1641
1642         if (!(client = pa_idxset_get_by_index(c->protocol->core->clients, index))) {
1643             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1644             return;
1645         }
1646
1647         pa_client_kill(client);
1648     } else if (command == PA_COMMAND_KILL_SINK_INPUT) {
1649         struct pa_sink_input *s;
1650
1651         if (!(s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index))) {
1652             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1653             return;
1654         }
1655
1656         pa_sink_input_kill(s);
1657     } else {
1658         struct pa_source_output *s;
1659
1660         assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
1661
1662         if (!(s = pa_idxset_get_by_index(c->protocol->core->source_outputs, index))) {
1663             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1664             return;
1665         }
1666
1667         pa_source_output_kill(s);
1668     }
1669
1670     pa_pstream_send_simple_ack(c->pstream, tag);
1671 }
1672
1673 static void command_load_module(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1674     struct connection *c = userdata;
1675     struct pa_module *m;
1676     const char *name, *argument;
1677     struct pa_tagstruct *reply;
1678     assert(c && t);
1679
1680     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
1681         pa_tagstruct_gets(t, &argument) < 0 ||
1682         !pa_tagstruct_eof(t)) {
1683         protocol_error(c);
1684         return;
1685     }
1686
1687     if (!c->authorized) {
1688         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1689         return;
1690     }
1691
1692     if (!(m = pa_module_load(c->protocol->core, name, argument))) {
1693         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INITFAILED);
1694         return;
1695     }
1696
1697     reply = pa_tagstruct_new(NULL, 0);
1698     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1699     pa_tagstruct_putu32(reply, tag);
1700     pa_tagstruct_putu32(reply, m->index);
1701     pa_pstream_send_tagstruct(c->pstream, reply);
1702 }
1703
1704 static void command_unload_module(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1705     struct connection *c = userdata;
1706     uint32_t index;
1707     struct pa_module *m;
1708     assert(c && t);
1709
1710     if (pa_tagstruct_getu32(t, &index) < 0 ||
1711         !pa_tagstruct_eof(t)) {
1712         protocol_error(c);
1713         return;
1714     }
1715
1716     if (!c->authorized) {
1717         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1718         return;
1719     }
1720
1721     if (!(m = pa_idxset_get_by_index(c->protocol->core->modules, index))) {
1722         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1723         return;
1724     }
1725
1726     pa_module_unload_request(m);
1727     pa_pstream_send_simple_ack(c->pstream, tag);
1728 }
1729
1730 static void command_add_autoload(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1731     struct connection *c = userdata;
1732     const char *name, *module, *argument;
1733     uint32_t type;
1734     uint32_t index;
1735     struct pa_tagstruct *reply;
1736     assert(c && t);
1737
1738     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
1739         pa_tagstruct_getu32(t, &type) < 0 || type > 1 ||
1740         pa_tagstruct_gets(t, &module) < 0 || !module ||
1741         pa_tagstruct_gets(t, &argument) < 0 ||
1742         !pa_tagstruct_eof(t)) {
1743         protocol_error(c);
1744         return;
1745     }
1746
1747     if (!c->authorized) {
1748         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1749         return;
1750     }
1751
1752     if (pa_autoload_add(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE, module, argument, &index) < 0) {
1753         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
1754         return;
1755     }
1756
1757     reply = pa_tagstruct_new(NULL, 0);
1758     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1759     pa_tagstruct_putu32(reply, tag);
1760     pa_tagstruct_putu32(reply, index);
1761     pa_pstream_send_tagstruct(c->pstream, reply);
1762 }
1763
1764 static void command_remove_autoload(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1765     struct connection *c = userdata;
1766     const char *name = NULL;
1767     uint32_t type, index = PA_IDXSET_INVALID;
1768     int r;
1769     assert(c && t);
1770
1771     if ((pa_tagstruct_getu32(t, &index) < 0 &&
1772         (pa_tagstruct_gets(t, &name) < 0 ||
1773          pa_tagstruct_getu32(t, &type) < 0)) ||
1774         (!name && index == PA_IDXSET_INVALID) ||
1775         (name && type > 1) ||
1776         !pa_tagstruct_eof(t)) {
1777         protocol_error(c);
1778         return;
1779     }
1780
1781     if (!c->authorized) {
1782         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1783         return;
1784     }
1785
1786     if (name)
1787         r = pa_autoload_remove_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
1788     else
1789         r = pa_autoload_remove_by_index(c->protocol->core, index);
1790
1791     if (r < 0) {
1792         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1793         return;
1794     }
1795
1796     pa_pstream_send_simple_ack(c->pstream, tag);
1797 }
1798
1799 static void autoload_fill_tagstruct(struct pa_tagstruct *t, const struct pa_autoload_entry *e) {
1800     assert(t && e);
1801
1802     pa_tagstruct_putu32(t, e->index);
1803     pa_tagstruct_puts(t, e->name);
1804     pa_tagstruct_putu32(t, e->type == PA_NAMEREG_SINK ? 0 : 1);
1805     pa_tagstruct_puts(t, e->module);
1806     pa_tagstruct_puts(t, e->argument);
1807 }
1808
1809 static void command_get_autoload_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1810     struct connection *c = userdata;
1811     const struct pa_autoload_entry *a = NULL;
1812     uint32_t type, index;
1813     const char *name;
1814     struct pa_tagstruct *reply;
1815     assert(c && t);
1816
1817     if ((pa_tagstruct_getu32(t, &index) < 0 &&
1818         (pa_tagstruct_gets(t, &name) < 0 ||
1819          pa_tagstruct_getu32(t, &type) < 0)) ||
1820         (!name && index == PA_IDXSET_INVALID) ||
1821         (name && type > 1) ||
1822         !pa_tagstruct_eof(t)) {
1823         protocol_error(c);
1824         return;
1825     }
1826
1827     if (!c->authorized) {
1828         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1829         return;
1830     }
1831
1832
1833     if (name)
1834         a = pa_autoload_get_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
1835     else
1836         a = pa_autoload_get_by_index(c->protocol->core, index);
1837
1838     if (!a) {
1839         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1840         return;
1841     }
1842
1843     reply = pa_tagstruct_new(NULL, 0);
1844     assert(reply);
1845     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1846     pa_tagstruct_putu32(reply, tag);
1847     autoload_fill_tagstruct(reply, a);
1848     pa_pstream_send_tagstruct(c->pstream, reply);
1849 }
1850
1851 static void command_get_autoload_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1852     struct connection *c = userdata;
1853     struct pa_tagstruct *reply;
1854     assert(c && t);
1855
1856     if (!pa_tagstruct_eof(t)) {
1857         protocol_error(c);
1858         return;
1859     }
1860
1861     if (!c->authorized) {
1862         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1863         return;
1864     }
1865
1866     reply = pa_tagstruct_new(NULL, 0);
1867     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1868     pa_tagstruct_putu32(reply, tag);
1869
1870     if (c->protocol->core->autoload_hashmap) {
1871         struct pa_autoload_entry *a;
1872         void *state = NULL;
1873
1874         while ((a = pa_hashmap_iterate(c->protocol->core->autoload_hashmap, &state, NULL)))
1875             autoload_fill_tagstruct(reply, a);
1876     }
1877
1878     pa_pstream_send_tagstruct(c->pstream, reply);
1879 }
1880
1881 /*** pstream callbacks ***/
1882
1883 static void pstream_packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
1884     struct connection *c = userdata;
1885     assert(p && packet && packet->data && c);
1886
1887     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
1888         pa_log(__FILE__": invalid packet.\n");
1889         connection_free(c);
1890     }
1891 }
1892
1893 static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, uint32_t delta, const struct pa_memchunk *chunk, void *userdata) {
1894     struct connection *c = userdata;
1895     struct output_stream *stream;
1896     assert(p && chunk && userdata);
1897
1898     if (!(stream = pa_idxset_get_by_index(c->output_streams, channel))) {
1899         pa_log(__FILE__": client sent block for invalid stream.\n");
1900         connection_free(c);
1901         return;
1902     }
1903
1904     if (stream->type == PLAYBACK_STREAM) {
1905         struct playback_stream *p = (struct playback_stream*) stream;
1906         if (chunk->length >= p->requested_bytes)
1907             p->requested_bytes = 0;
1908         else
1909             p->requested_bytes -= chunk->length;
1910
1911         pa_memblockq_push_align(p->memblockq, chunk, delta);
1912         assert(p->sink_input);
1913 /*         pa_log(__FILE__": after_recv: %u\n", pa_memblockq_get_length(p->memblockq)); */
1914
1915         pa_sink_notify(p->sink_input->sink);
1916 /*          pa_log(__FILE__": Recieved %u bytes.\n", chunk->length);  */
1917
1918     } else {
1919         struct upload_stream *u = (struct upload_stream*) stream;
1920         size_t l;
1921         assert(u->type == UPLOAD_STREAM);
1922
1923         if (!u->memchunk.memblock) {
1924             if (u->length == chunk->length) {
1925                 u->memchunk = *chunk;
1926                 pa_memblock_ref(u->memchunk.memblock);
1927                 u->length = 0;
1928             } else {
1929                 u->memchunk.memblock = pa_memblock_new(u->length, c->protocol->core->memblock_stat);
1930                 u->memchunk.index = u->memchunk.length = 0;
1931             }
1932         }
1933
1934         assert(u->memchunk.memblock);
1935
1936         l = u->length;
1937         if (l > chunk->length)
1938             l = chunk->length;
1939
1940         if (l > 0) {
1941             memcpy((uint8_t*) u->memchunk.memblock->data + u->memchunk.index + u->memchunk.length,
1942                    (uint8_t*) chunk->memblock->data+chunk->index, l);
1943             u->memchunk.length += l;
1944             u->length -= l;
1945         }
1946     }
1947 }
1948
1949 static void pstream_die_callback(struct pa_pstream *p, void *userdata) {
1950     struct connection *c = userdata;
1951     assert(p && c);
1952     connection_free(c);
1953
1954 /*    pa_log(__FILE__": connection died.\n");*/
1955 }
1956
1957
1958 static void pstream_drain_callback(struct pa_pstream *p, void *userdata) {
1959     struct connection *c = userdata;
1960     assert(p && c);
1961
1962     send_memblock(c);
1963 }
1964
1965 /*** client callbacks ***/
1966
1967 static void client_kill_cb(struct pa_client *c) {
1968     assert(c && c->userdata);
1969     connection_free(c->userdata);
1970 }
1971
1972 /*** socket server callbacks ***/
1973
1974 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
1975     struct pa_protocol_native *p = userdata;
1976     struct connection *c;
1977     assert(io && p);
1978
1979     c = pa_xmalloc(sizeof(struct connection));
1980
1981     c->authorized =!! p->public;
1982     c->protocol = p;
1983     assert(p->core);
1984     c->client = pa_client_new(p->core, "NATIVE", "Client");
1985     assert(c->client);
1986     c->client->kill = client_kill_cb;
1987     c->client->userdata = c;
1988     c->client->owner = p->module;
1989
1990     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->memblock_stat);
1991     assert(c->pstream);
1992
1993     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
1994     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
1995     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
1996     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
1997
1998     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
1999     assert(c->pdispatch);
2000
2001     c->record_streams = pa_idxset_new(NULL, NULL);
2002     c->output_streams = pa_idxset_new(NULL, NULL);
2003     assert(c->record_streams && c->output_streams);
2004
2005     c->rrobin_index = PA_IDXSET_INVALID;
2006     c->subscription = NULL;
2007
2008     pa_idxset_put(p->connections, c, NULL);
2009 }
2010
2011 /*** module entry points ***/
2012
2013 static int load_key(struct pa_protocol_native*p, const char*fn) {
2014     assert(p);
2015
2016     p->auth_cookie_in_property = 0;
2017
2018     if (!fn && pa_authkey_prop_get(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0) {
2019         pa_log(__FILE__": using already loaded auth cookie.\n");
2020         pa_authkey_prop_ref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
2021         p->auth_cookie_in_property = 1;
2022         return 0;
2023     }
2024
2025     if (!fn)
2026         fn = PA_NATIVE_COOKIE_FILE;
2027
2028     if (pa_authkey_load_from_home(fn, p->auth_cookie, sizeof(p->auth_cookie)) < 0)
2029         return -1;
2030
2031     pa_log(__FILE__": loading cookie from disk.\n");
2032
2033     if (pa_authkey_prop_put(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0)
2034         p->auth_cookie_in_property = 1;
2035
2036     return 0;
2037 }
2038
2039 static struct pa_protocol_native* protocol_new_internal(struct pa_core *c, struct pa_module *m, struct pa_modargs *ma) {
2040     struct pa_protocol_native *p;
2041     int public = 0;
2042     assert(c && ma);
2043
2044     if (pa_modargs_get_value_boolean(ma, "public", &public) < 0) {
2045         pa_log(__FILE__": public= expects a boolean argument.\n");
2046         return NULL;
2047     }
2048
2049     p = pa_xmalloc(sizeof(struct pa_protocol_native));
2050     p->core = c;
2051     p->module = m;
2052     p->public = public;
2053     p->server = NULL;
2054
2055     if (load_key(p, pa_modargs_get_value(ma, "cookie", NULL)) < 0) {
2056         pa_xfree(p);
2057         return NULL;
2058     }
2059
2060     p->connections = pa_idxset_new(NULL, NULL);
2061     assert(p->connections);
2062
2063     return p;
2064 }
2065
2066 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) {
2067     struct pa_protocol_native *p;
2068
2069     if (!(p = protocol_new_internal(core, m, ma)))
2070         return NULL;
2071
2072     p->server = server;
2073     pa_socket_server_set_callback(p->server, on_connection, p);
2074
2075     return p;
2076 }
2077
2078 void pa_protocol_native_free(struct pa_protocol_native *p) {
2079     struct connection *c;
2080     assert(p);
2081
2082     while ((c = pa_idxset_first(p->connections, NULL)))
2083         connection_free(c);
2084     pa_idxset_free(p->connections, NULL, NULL);
2085
2086     if (p->server)
2087         pa_socket_server_unref(p->server);
2088
2089     if (p->auth_cookie_in_property)
2090         pa_authkey_prop_unref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
2091
2092     pa_xfree(p);
2093 }
2094
2095 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) {
2096     struct pa_protocol_native *p;
2097
2098     if (!(p = protocol_new_internal(core, m, ma)))
2099         return NULL;
2100
2101     on_connection(NULL, io, p);
2102
2103     return p;
2104 }