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