src/pulsecore/protocol-native.c

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