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