code.delx.au - pulseaudio/blob - 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 <stdlib.h>
  32 #include <unistd.h>
  33
  34 #include <pulse/timeval.h>
  35 #include <pulse/version.h>
  36 #include <pulse/utf8.h>
  37 #include <pulse/util.h>
  38 #include <pulse/xmalloc.h>
  39
  40 #include <pulsecore/native-common.h>
  41 #include <pulsecore/packet.h>
  42 #include <pulsecore/client.h>
  43 #include <pulsecore/source-output.h>
  44 #include <pulsecore/sink-input.h>
  45 #include <pulsecore/pstream.h>
  46 #include <pulsecore/tagstruct.h>
  47 #include <pulsecore/pdispatch.h>
  48 #include <pulsecore/pstream-util.h>
  49 #include <pulsecore/authkey.h>
  50 #include <pulsecore/namereg.h>
  51 #include <pulsecore/core-scache.h>
  52 #include <pulsecore/core-subscribe.h>
  53 #include <pulsecore/log.h>
  54 #include <pulsecore/autoload.h>
  55 #include <pulsecore/authkey-prop.h>
  56 #include <pulsecore/strlist.h>
  57 #include <pulsecore/props.h>
  58 #include <pulsecore/sample-util.h>
  59 #include <pulsecore/llist.h>
  60 #include <pulsecore/creds.h>
  61 #include <pulsecore/core-util.h>
  62 #include <pulsecore/ipacl.h>
  63 #include <pulsecore/thread-mq.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 typedef struct connection connection;
  76 struct pa_protocol_native;
  77
  78 typedef struct record_stream {
  79     pa_msgobject parent;
  80
  81     connection *connection;
  82     uint32_t index;
  83
  84     pa_source_output *source_output;
  85     pa_memblockq *memblockq;
  86     size_t fragment_size;
  87 } record_stream;
  88
  89 typedef struct output_stream {
  90     pa_msgobject parent;
  91 } output_stream;
  92
  93 typedef struct playback_stream {
  94     output_stream parent;
  95
  96     connection *connection;
  97     uint32_t index;
  98
  99     pa_sink_input *sink_input;
 100     pa_memblockq *memblockq;
 101     pa_bool_t drain_request;
 102     uint32_t drain_tag;
 103     uint32_t syncid;
 104     pa_bool_t underrun;
 105
 106     pa_atomic_t missing;
 107     size_t minreq;
 108
 109     /* Only updated after SINK_INPUT_MESSAGE_UPDATE_LATENCY */
 110     int64_t read_index, write_index;
 111     size_t render_memblockq_length;
 112 } playback_stream;
 113
 114 typedef struct upload_stream {
 115     output_stream parent;
 116
 117     connection *connection;
 118     uint32_t index;
 119
 120     pa_memchunk memchunk;
 121     size_t length;
 122     char *name;
 123     pa_sample_spec sample_spec;
 124     pa_channel_map channel_map;
 125     pa_proplist *proplist;
 126 } upload_stream;
 127
 128 struct connection {
 129     pa_msgobject parent;
 130
 131     pa_bool_t authorized;
 132     uint32_t version;
 133     pa_protocol_native *protocol;
 134     pa_client *client;
 135     pa_pstream *pstream;
 136     pa_pdispatch *pdispatch;
 137     pa_idxset *record_streams, *output_streams;
 138     uint32_t rrobin_index;
 139     pa_subscription *subscription;
 140     pa_time_event *auth_timeout_event;
 141 };
 142
 143 PA_DECLARE_CLASS(record_stream);
 144 #define RECORD_STREAM(o) (record_stream_cast(o))
 145 static PA_DEFINE_CHECK_TYPE(record_stream, pa_msgobject);
 146
 147 PA_DECLARE_CLASS(output_stream);
 148 #define OUTPUT_STREAM(o) (output_stream_cast(o))
 149 static PA_DEFINE_CHECK_TYPE(output_stream, pa_msgobject);
 150
 151 PA_DECLARE_CLASS(playback_stream);
 152 #define PLAYBACK_STREAM(o) (playback_stream_cast(o))
 153 static PA_DEFINE_CHECK_TYPE(playback_stream, output_stream);
 154
 155 PA_DECLARE_CLASS(upload_stream);
 156 #define UPLOAD_STREAM(o) (upload_stream_cast(o))
 157 static PA_DEFINE_CHECK_TYPE(upload_stream, output_stream);
 158
 159 PA_DECLARE_CLASS(connection);
 160 #define CONNECTION(o) (connection_cast(o))
 161 static PA_DEFINE_CHECK_TYPE(connection, pa_msgobject);
 162
 163 struct pa_protocol_native {
 164     pa_module *module;
 165     pa_core *core;
 166     pa_bool_t public;
 167     pa_socket_server *server;
 168     pa_idxset *connections;
 169     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
 170     pa_bool_t auth_cookie_in_property;
 171 #ifdef HAVE_CREDS
 172     char *auth_group;
 173 #endif
 174     pa_ip_acl *auth_ip_acl;
 175 };
 176
 177 enum {
 178     SINK_INPUT_MESSAGE_POST_DATA = PA_SINK_INPUT_MESSAGE_MAX, /* data from main loop to sink input */
 179     SINK_INPUT_MESSAGE_DRAIN, /* disabled prebuf, get playback started. */
 180     SINK_INPUT_MESSAGE_FLUSH,
 181     SINK_INPUT_MESSAGE_TRIGGER,
 182     SINK_INPUT_MESSAGE_SEEK,
 183     SINK_INPUT_MESSAGE_PREBUF_FORCE,
 184     SINK_INPUT_MESSAGE_UPDATE_LATENCY
 185 };
 186
 187 enum {
 188     PLAYBACK_STREAM_MESSAGE_REQUEST_DATA,      /* data requested from sink input from the main loop */
 189     PLAYBACK_STREAM_MESSAGE_UNDERFLOW,
 190     PLAYBACK_STREAM_MESSAGE_OVERFLOW,
 191     PLAYBACK_STREAM_MESSAGE_DRAIN_ACK
 192 };
 193
 194 enum {
 195     RECORD_STREAM_MESSAGE_POST_DATA         /* data from source output to main loop */
 196 };
 197
 198 enum {
 199     CONNECTION_MESSAGE_RELEASE,
 200     CONNECTION_MESSAGE_REVOKE
 201 };
 202
 203 static int sink_input_pop_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk);
 204 static void sink_input_kill_cb(pa_sink_input *i);
 205 static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspend);
 206 static void sink_input_moved_cb(pa_sink_input *i);
 207
 208 static void send_memblock(connection *c);
 209 static void request_bytes(struct playback_stream*s);
 210
 211 static void source_output_kill_cb(pa_source_output *o);
 212 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk);
 213 static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend);
 214 static void source_output_moved_cb(pa_source_output *o);
 215 static pa_usec_t source_output_get_latency_cb(pa_source_output *o);
 216
 217 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
 218
 219 static void command_exit(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 220 static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 221 static void command_drain_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 222 static void command_create_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 223 static void command_delete_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 224 static void command_auth(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 225 static void command_set_client_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 226 static void command_lookup(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 227 static void command_stat(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 228 static void command_get_playback_latency(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 229 static void command_get_record_latency(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 230 static void command_create_upload_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 231 static void command_finish_upload_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 232 static void command_play_sample(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 233 static void command_remove_sample(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 234 static void command_get_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 235 static void command_get_info_list(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 236 static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 237 static void command_subscribe(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 238 static void command_set_volume(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 239 static void command_set_mute(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 240 static void command_cork_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 241 static void command_trigger_or_flush_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 242 static void command_set_default_sink_or_source(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 243 static void command_set_stream_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 244 static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 245 static void command_load_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 246 static void command_unload_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 247 static void command_add_autoload(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 248 static void command_remove_autoload(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 249 static void command_get_autoload_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 250 static void command_get_autoload_info_list(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 251 static void command_cork_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 252 static void command_flush_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 253 static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 254 static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 255 static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 256 static void command_update_stream_sample_rate(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 257 static void command_update_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 258 static void command_remove_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 259
 260 static const pa_pdispatch_cb_t command_table[PA_COMMAND_MAX] = {
 261     [PA_COMMAND_ERROR] = NULL,
 262     [PA_COMMAND_TIMEOUT] = NULL,
 263     [PA_COMMAND_REPLY] = NULL,
 264     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = command_create_playback_stream,
 265     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = command_delete_stream,
 266     [PA_COMMAND_DRAIN_PLAYBACK_STREAM] = command_drain_playback_stream,
 267     [PA_COMMAND_CREATE_RECORD_STREAM] = command_create_record_stream,
 268     [PA_COMMAND_DELETE_RECORD_STREAM] = command_delete_stream,
 269     [PA_COMMAND_AUTH] = command_auth,
 270     [PA_COMMAND_REQUEST] = NULL,
 271     [PA_COMMAND_EXIT] = command_exit,
 272     [PA_COMMAND_SET_CLIENT_NAME] = command_set_client_name,
 273     [PA_COMMAND_LOOKUP_SINK] = command_lookup,
 274     [PA_COMMAND_LOOKUP_SOURCE] = command_lookup,
 275     [PA_COMMAND_STAT] = command_stat,
 276     [PA_COMMAND_GET_PLAYBACK_LATENCY] = command_get_playback_latency,
 277     [PA_COMMAND_GET_RECORD_LATENCY] = command_get_record_latency,
 278     [PA_COMMAND_CREATE_UPLOAD_STREAM] = command_create_upload_stream,
 279     [PA_COMMAND_DELETE_UPLOAD_STREAM] = command_delete_stream,
 280     [PA_COMMAND_FINISH_UPLOAD_STREAM] = command_finish_upload_stream,
 281     [PA_COMMAND_PLAY_SAMPLE] = command_play_sample,
 282     [PA_COMMAND_REMOVE_SAMPLE] = command_remove_sample,
 283     [PA_COMMAND_GET_SINK_INFO] = command_get_info,
 284     [PA_COMMAND_GET_SOURCE_INFO] = command_get_info,
 285     [PA_COMMAND_GET_CLIENT_INFO] = command_get_info,
 286     [PA_COMMAND_GET_MODULE_INFO] = command_get_info,
 287     [PA_COMMAND_GET_SINK_INPUT_INFO] = command_get_info,
 288     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO] = command_get_info,
 289     [PA_COMMAND_GET_SAMPLE_INFO] = command_get_info,
 290     [PA_COMMAND_GET_SINK_INFO_LIST] = command_get_info_list,
 291     [PA_COMMAND_GET_SOURCE_INFO_LIST] = command_get_info_list,
 292     [PA_COMMAND_GET_MODULE_INFO_LIST] = command_get_info_list,
 293     [PA_COMMAND_GET_CLIENT_INFO_LIST] = command_get_info_list,
 294     [PA_COMMAND_GET_SINK_INPUT_INFO_LIST] = command_get_info_list,
 295     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST] = command_get_info_list,
 296     [PA_COMMAND_GET_SAMPLE_INFO_LIST] = command_get_info_list,
 297     [PA_COMMAND_GET_SERVER_INFO] = command_get_server_info,
 298     [PA_COMMAND_SUBSCRIBE] = command_subscribe,
 299
 300     [PA_COMMAND_SET_SINK_VOLUME] = command_set_volume,
 301     [PA_COMMAND_SET_SINK_INPUT_VOLUME] = command_set_volume,
 302     [PA_COMMAND_SET_SOURCE_VOLUME] = command_set_volume,
 303
 304     [PA_COMMAND_SET_SINK_MUTE] = command_set_mute,
 305     [PA_COMMAND_SET_SINK_INPUT_MUTE] = command_set_mute,
 306     [PA_COMMAND_SET_SOURCE_MUTE] = command_set_mute,
 307
 308     [PA_COMMAND_SUSPEND_SINK] = command_suspend,
 309     [PA_COMMAND_SUSPEND_SOURCE] = command_suspend,
 310
 311     [PA_COMMAND_CORK_PLAYBACK_STREAM] = command_cork_playback_stream,
 312     [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
 313     [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
 314     [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
 315
 316     [PA_COMMAND_CORK_RECORD_STREAM] = command_cork_record_stream,
 317     [PA_COMMAND_FLUSH_RECORD_STREAM] = command_flush_record_stream,
 318
 319     [PA_COMMAND_SET_DEFAULT_SINK] = command_set_default_sink_or_source,
 320     [PA_COMMAND_SET_DEFAULT_SOURCE] = command_set_default_sink_or_source,
 321     [PA_COMMAND_SET_PLAYBACK_STREAM_NAME] = command_set_stream_name,
 322     [PA_COMMAND_SET_RECORD_STREAM_NAME] = command_set_stream_name,
 323     [PA_COMMAND_KILL_CLIENT] = command_kill,
 324     [PA_COMMAND_KILL_SINK_INPUT] = command_kill,
 325     [PA_COMMAND_KILL_SOURCE_OUTPUT] = command_kill,
 326     [PA_COMMAND_LOAD_MODULE] = command_load_module,
 327     [PA_COMMAND_UNLOAD_MODULE] = command_unload_module,
 328     [PA_COMMAND_GET_AUTOLOAD_INFO] = command_get_autoload_info,
 329     [PA_COMMAND_GET_AUTOLOAD_INFO_LIST] = command_get_autoload_info_list,
 330     [PA_COMMAND_ADD_AUTOLOAD] = command_add_autoload,
 331     [PA_COMMAND_REMOVE_AUTOLOAD] = command_remove_autoload,
 332
 333     [PA_COMMAND_MOVE_SINK_INPUT] = command_move_stream,
 334     [PA_COMMAND_MOVE_SOURCE_OUTPUT] = command_move_stream,
 335
 336     [PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR] = command_set_stream_buffer_attr,
 337     [PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR] = command_set_stream_buffer_attr,
 338
 339     [PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE] = command_update_stream_sample_rate,
 340     [PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE] = command_update_stream_sample_rate,
 341
 342     [PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST] = command_update_proplist,
 343     [PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST] = command_update_proplist,
 344     [PA_COMMAND_UPDATE_CLIENT_PROPLIST] = command_update_proplist,
 345
 346     [PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST] = command_remove_proplist,
 347     [PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST] = command_remove_proplist,
 348     [PA_COMMAND_REMOVE_CLIENT_PROPLIST] = command_remove_proplist,
 349 };
 350
 351 /* structure management */
 352
 353 static void upload_stream_unlink(upload_stream *s) {
 354     pa_assert(s);
 355
 356     if (!s->connection)
 357         return;
 358
 359     pa_assert_se(pa_idxset_remove_by_data(s->connection->output_streams, s, NULL) == s);
 360     s->connection = NULL;
 361     upload_stream_unref(s);
 362 }
 363
 364 static void upload_stream_free(pa_object *o) {
 365     upload_stream *s = UPLOAD_STREAM(o);
 366     pa_assert(s);
 367
 368     upload_stream_unlink(s);
 369
 370     pa_xfree(s->name);
 371
 372     if (s->proplist)
 373         pa_proplist_free(s->proplist);
 374
 375     if (s->memchunk.memblock)
 376         pa_memblock_unref(s->memchunk.memblock);
 377
 378     pa_xfree(s);
 379 }
 380
 381 static upload_stream* upload_stream_new(
 382         connection *c,
 383         const pa_sample_spec *ss,
 384         const pa_channel_map *map,
 385         const char *name,
 386         size_t length,
 387         pa_proplist *p) {
 388
 389     upload_stream *s;
 390
 391     pa_assert(c);
 392     pa_assert(ss);
 393     pa_assert(name);
 394     pa_assert(length > 0);
 395     pa_assert(p);
 396
 397     s = pa_msgobject_new(upload_stream);
 398     s->parent.parent.parent.free = upload_stream_free;
 399     s->connection = c;
 400     s->sample_spec = *ss;
 401     s->channel_map = *map;
 402     s->name = pa_xstrdup(name);
 403     pa_memchunk_reset(&s->memchunk);
 404     s->length = length;
 405     s->proplist = pa_proplist_copy(p);
 406     pa_proplist_update(s->proplist, PA_UPDATE_MERGE, c->client->proplist);
 407
 408     pa_idxset_put(c->output_streams, s, &s->index);
 409
 410     return s;
 411 }
 412
 413 static void record_stream_unlink(record_stream *s) {
 414     pa_assert(s);
 415
 416     if (!s->connection)
 417         return;
 418
 419     if (s->source_output) {
 420         pa_source_output_unlink(s->source_output);
 421         pa_source_output_unref(s->source_output);
 422         s->source_output = NULL;
 423     }
 424
 425     pa_assert_se(pa_idxset_remove_by_data(s->connection->record_streams, s, NULL) == s);
 426     s->connection = NULL;
 427     record_stream_unref(s);
 428 }
 429
 430 static void record_stream_free(pa_object *o) {
 431     record_stream *s = RECORD_STREAM(o);
 432     pa_assert(s);
 433
 434     record_stream_unlink(s);
 435
 436     pa_memblockq_free(s->memblockq);
 437     pa_xfree(s);
 438 }
 439
 440 static int record_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 441     record_stream *s = RECORD_STREAM(o);
 442     record_stream_assert_ref(s);
 443
 444     if (!s->connection)
 445         return -1;
 446
 447     switch (code) {
 448
 449         case RECORD_STREAM_MESSAGE_POST_DATA:
 450
 451             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
 452 /*                 pa_log_warn("Failed to push data into output queue."); */
 453                 return -1;
 454             }
 455
 456             if (!pa_pstream_is_pending(s->connection->pstream))
 457                 send_memblock(s->connection);
 458
 459             break;
 460     }
 461
 462     return 0;
 463 }
 464
 465 static record_stream* record_stream_new(
 466         connection *c,
 467         pa_source *source,
 468         pa_sample_spec *ss,
 469         pa_channel_map *map,
 470         const char *name,
 471         uint32_t *maxlength,
 472         uint32_t fragment_size,
 473         pa_source_output_flags_t flags,
 474         pa_proplist *p) {
 475
 476     record_stream *s;
 477     pa_source_output *source_output;
 478     size_t base;
 479     pa_source_output_new_data data;
 480
 481     pa_assert(c);
 482     pa_assert(ss);
 483     pa_assert(name);
 484     pa_assert(maxlength);
 485     pa_assert(*maxlength > 0);
 486     pa_assert(p);
 487
 488     pa_source_output_new_data_init(&data);
 489
 490     pa_proplist_update(data.proplist, PA_UPDATE_REPLACE, p);
 491     pa_proplist_update(data.proplist, PA_UPDATE_MERGE, c->client->proplist);
 492     data.driver = __FILE__;
 493     data.module = c->protocol->module;
 494     data.client = c->client;
 495     data.source = source;
 496     pa_source_output_new_data_set_sample_spec(&data, ss);
 497     pa_source_output_new_data_set_channel_map(&data, map);
 498
 499     source_output = pa_source_output_new(c->protocol->core, &data, flags);
 500
 501     pa_source_output_new_data_done(&data);
 502
 503     if (!source_output)
 504         return NULL;
 505
 506     s = pa_msgobject_new(record_stream);
 507     s->parent.parent.free = record_stream_free;
 508     s->parent.process_msg = record_stream_process_msg;
 509     s->connection = c;
 510     s->source_output = source_output;
 511     s->source_output->push = source_output_push_cb;
 512     s->source_output->kill = source_output_kill_cb;
 513     s->source_output->get_latency = source_output_get_latency_cb;
 514     s->source_output->moved = source_output_moved_cb;
 515     s->source_output->suspend = source_output_suspend_cb;
 516     s->source_output->userdata = s;
 517
 518     s->memblockq = pa_memblockq_new(
 519             0,
 520             *maxlength,
 521             0,
 522             base = pa_frame_size(&s->source_output->sample_spec),
 523             1,
 524             0,
 525             0,
 526             NULL);
 527
 528     *maxlength = pa_memblockq_get_maxlength(s->memblockq);
 529
 530     s->fragment_size = (fragment_size/base)*base;
 531     if (s->fragment_size <= 0)
 532         s->fragment_size = base;
 533
 534     if (s->fragment_size > *maxlength)
 535         s->fragment_size = *maxlength;
 536
 537     *ss = s->source_output->sample_spec;
 538     *map = s->source_output->channel_map;
 539
 540     pa_idxset_put(c->record_streams, s, &s->index);
 541
 542     pa_source_output_put(s->source_output);
 543     return s;
 544 }
 545
 546 static void playback_stream_unlink(playback_stream *s) {
 547     pa_assert(s);
 548
 549     if (!s->connection)
 550         return;
 551
 552     if (s->sink_input) {
 553         pa_sink_input_unlink(s->sink_input);
 554         pa_sink_input_unref(s->sink_input);
 555         s->sink_input = NULL;
 556     }
 557
 558     if (s->drain_request)
 559         pa_pstream_send_error(s->connection->pstream, s->drain_tag, PA_ERR_NOENTITY);
 560
 561     pa_assert_se(pa_idxset_remove_by_data(s->connection->output_streams, s, NULL) == s);
 562     s->connection = NULL;
 563     playback_stream_unref(s);
 564 }
 565
 566 static void playback_stream_free(pa_object* o) {
 567     playback_stream *s = PLAYBACK_STREAM(o);
 568     pa_assert(s);
 569
 570     playback_stream_unlink(s);
 571
 572     pa_memblockq_free(s->memblockq);
 573     pa_xfree(s);
 574 }
 575
 576 static int playback_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 577     playback_stream *s = PLAYBACK_STREAM(o);
 578     playback_stream_assert_ref(s);
 579
 580     if (!s->connection)
 581         return -1;
 582
 583     switch (code) {
 584         case PLAYBACK_STREAM_MESSAGE_REQUEST_DATA: {
 585             pa_tagstruct *t;
 586             uint32_t l = 0;
 587
 588             for (;;) {
 589                 int32_t k;
 590
 591                 if ((k = pa_atomic_load(&s->missing)) <= 0)
 592                     break;
 593
 594                 l += k;
 595
 596                 if (l < s->minreq)
 597                     break;
 598
 599                 if (pa_atomic_sub(&s->missing, k) <= k)
 600                     break;
 601             }
 602
 603             if (l < s->minreq)
 604                 break;
 605
 606             t = pa_tagstruct_new(NULL, 0);
 607             pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 608             pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 609             pa_tagstruct_putu32(t, s->index);
 610             pa_tagstruct_putu32(t, l);
 611             pa_pstream_send_tagstruct(s->connection->pstream, t);
 612
 613 /*             pa_log("Requesting %u bytes", l);     */
 614             break;
 615         }
 616
 617         case PLAYBACK_STREAM_MESSAGE_UNDERFLOW: {
 618             pa_tagstruct *t;
 619
 620             /* Report that we're empty */
 621             t = pa_tagstruct_new(NULL, 0);
 622             pa_tagstruct_putu32(t, PA_COMMAND_UNDERFLOW);
 623             pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 624             pa_tagstruct_putu32(t, s->index);
 625             pa_pstream_send_tagstruct(s->connection->pstream, t);
 626             break;
 627         }
 628
 629         case PLAYBACK_STREAM_MESSAGE_OVERFLOW: {
 630             pa_tagstruct *t;
 631
 632             /* Notify the user we're overflowed*/
 633             t = pa_tagstruct_new(NULL, 0);
 634             pa_tagstruct_putu32(t, PA_COMMAND_OVERFLOW);
 635             pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 636             pa_tagstruct_putu32(t, s->index);
 637             pa_pstream_send_tagstruct(s->connection->pstream, t);
 638             break;
 639         }
 640
 641         case PLAYBACK_STREAM_MESSAGE_DRAIN_ACK:
 642             pa_pstream_send_simple_ack(s->connection->pstream, PA_PTR_TO_UINT(userdata));
 643             break;
 644
 645     }
 646
 647     return 0;
 648 }
 649
 650 static playback_stream* playback_stream_new(
 651         connection *c,
 652         pa_sink *sink,
 653         pa_sample_spec *ss,
 654         pa_channel_map *map,
 655         const char *name,
 656         uint32_t *maxlength,
 657         uint32_t *tlength,
 658         uint32_t *prebuf,
 659         uint32_t *minreq,
 660         pa_cvolume *volume,
 661         uint32_t syncid,
 662         uint32_t *missing,
 663         pa_sink_input_flags_t flags,
 664         pa_proplist *p) {
 665
 666     playback_stream *s, *ssync;
 667     pa_sink_input *sink_input;
 668     pa_memblock *silence;
 669     uint32_t idx;
 670     int64_t start_index;
 671     pa_sink_input_new_data data;
 672
 673     pa_assert(c);
 674     pa_assert(ss);
 675     pa_assert(name);
 676     pa_assert(maxlength);
 677     pa_assert(p);
 678
 679     /* Find syncid group */
 680     for (ssync = pa_idxset_first(c->output_streams, &idx); ssync; ssync = pa_idxset_next(c->output_streams, &idx)) {
 681
 682         if (!playback_stream_isinstance(ssync))
 683             continue;
 684
 685         if (ssync->syncid == syncid)
 686             break;
 687     }
 688
 689     /* Synced streams must connect to the same sink */
 690     if (ssync) {
 691
 692         if (!sink)
 693             sink = ssync->sink_input->sink;
 694         else if (sink != ssync->sink_input->sink)
 695             return NULL;
 696     }
 697
 698     pa_sink_input_new_data_init(&data);
 699
 700     pa_proplist_update(data.proplist, PA_UPDATE_REPLACE, p);
 701     pa_proplist_update(data.proplist, PA_UPDATE_MERGE, c->client->proplist);
 702     data.driver = __FILE__;
 703     data.module = c->protocol->module;
 704     data.client = c->client;
 705     data.sink = sink;
 706     pa_sink_input_new_data_set_sample_spec(&data, ss);
 707     pa_sink_input_new_data_set_channel_map(&data, map);
 708     pa_sink_input_new_data_set_volume(&data, volume);
 709     data.sync_base = ssync ? ssync->sink_input : NULL;
 710
 711     sink_input = pa_sink_input_new(c->protocol->core, &data, flags);
 712
 713     pa_sink_input_new_data_done(&data);
 714
 715     if (!sink_input)
 716         return NULL;
 717
 718     s = pa_msgobject_new(playback_stream);
 719     s->parent.parent.parent.free = playback_stream_free;
 720     s->parent.parent.process_msg = playback_stream_process_msg;
 721     s->connection = c;
 722     s->syncid = syncid;
 723     s->sink_input = sink_input;
 724     s->underrun = TRUE;
 725
 726     s->sink_input->parent.process_msg = sink_input_process_msg;
 727     s->sink_input->pop = sink_input_pop_cb;
 728     s->sink_input->kill = sink_input_kill_cb;
 729     s->sink_input->moved = sink_input_moved_cb;
 730     s->sink_input->suspend = sink_input_suspend_cb;
 731     s->sink_input->userdata = s;
 732
 733     start_index = ssync ? pa_memblockq_get_read_index(ssync->memblockq) : 0;
 734
 735     silence = pa_silence_memblock_new(c->protocol->core->mempool, &s->sink_input->sample_spec, 0);
 736
 737     s->memblockq = pa_memblockq_new(
 738             start_index,
 739             *maxlength,
 740             *tlength,
 741             pa_frame_size(&s->sink_input->sample_spec),
 742             *prebuf,
 743             *minreq,
 744             0,
 745             silence);
 746
 747     pa_memblock_unref(silence);
 748
 749     *maxlength = (uint32_t) pa_memblockq_get_maxlength(s->memblockq);
 750     *tlength = (uint32_t) pa_memblockq_get_tlength(s->memblockq);
 751     *prebuf = (uint32_t) pa_memblockq_get_prebuf(s->memblockq);
 752     *minreq = (uint32_t) pa_memblockq_get_minreq(s->memblockq);
 753     *missing = (uint32_t) pa_memblockq_pop_missing(s->memblockq);
 754
 755     *ss = s->sink_input->sample_spec;
 756     *map = s->sink_input->channel_map;
 757
 758     s->minreq = pa_memblockq_get_minreq(s->memblockq);
 759     pa_atomic_store(&s->missing, 0);
 760     s->drain_request = FALSE;
 761
 762     pa_idxset_put(c->output_streams, s, &s->index);
 763
 764     pa_sink_input_put(s->sink_input);
 765
 766     return s;
 767 }
 768
 769 static int connection_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 770     connection *c = CONNECTION(o);
 771     connection_assert_ref(c);
 772
 773     if (!c->protocol)
 774         return -1;
 775
 776     switch (code) {
 777
 778         case CONNECTION_MESSAGE_REVOKE:
 779             pa_pstream_send_revoke(c->pstream, PA_PTR_TO_UINT(userdata));
 780             break;
 781
 782         case CONNECTION_MESSAGE_RELEASE:
 783             pa_pstream_send_release(c->pstream, PA_PTR_TO_UINT(userdata));
 784             break;
 785     }
 786
 787     return 0;
 788 }
 789
 790 static void connection_unlink(connection *c) {
 791     record_stream *r;
 792     output_stream *o;
 793
 794     pa_assert(c);
 795
 796     if (!c->protocol)
 797         return;
 798
 799     while ((r = pa_idxset_first(c->record_streams, NULL)))
 800         record_stream_unlink(r);
 801
 802     while ((o = pa_idxset_first(c->output_streams, NULL)))
 803         if (playback_stream_isinstance(o))
 804             playback_stream_unlink(PLAYBACK_STREAM(o));
 805         else
 806             upload_stream_unlink(UPLOAD_STREAM(o));
 807
 808     if (c->subscription)
 809         pa_subscription_free(c->subscription);
 810
 811     if (c->pstream)
 812         pa_pstream_unlink(c->pstream);
 813
 814     if (c->auth_timeout_event) {
 815         c->protocol->core->mainloop->time_free(c->auth_timeout_event);
 816         c->auth_timeout_event = NULL;
 817     }
 818
 819     pa_assert_se(pa_idxset_remove_by_data(c->protocol->connections, c, NULL) == c);
 820     c->protocol = NULL;
 821     connection_unref(c);
 822 }
 823
 824 static void connection_free(pa_object *o) {
 825     connection *c = CONNECTION(o);
 826
 827     pa_assert(c);
 828
 829     connection_unlink(c);
 830
 831     pa_idxset_free(c->record_streams, NULL, NULL);
 832     pa_idxset_free(c->output_streams, NULL, NULL);
 833
 834     pa_pdispatch_unref(c->pdispatch);
 835     pa_pstream_unref(c->pstream);
 836     pa_client_free(c->client);
 837
 838     pa_xfree(c);
 839 }
 840
 841 /* Called from thread context */
 842 static void request_bytes(playback_stream *s) {
 843     size_t m, previous_missing;
 844
 845     playback_stream_assert_ref(s);
 846
 847     m = pa_memblockq_pop_missing(s->memblockq);
 848
 849     if (m <= 0)
 850         return;
 851
 852 /*     pa_log("request_bytes(%u)", m); */
 853
 854     previous_missing = pa_atomic_add(&s->missing, m);
 855     if (previous_missing < s->minreq && previous_missing+m >= s->minreq) {
 856         pa_assert(pa_thread_mq_get());
 857         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
 858     }
 859 }
 860
 861 static void send_memblock(connection *c) {
 862     uint32_t start;
 863     record_stream *r;
 864
 865     start = PA_IDXSET_INVALID;
 866     for (;;) {
 867         pa_memchunk chunk;
 868
 869         if (!(r = RECORD_STREAM(pa_idxset_rrobin(c->record_streams, &c->rrobin_index))))
 870             return;
 871
 872         if (start == PA_IDXSET_INVALID)
 873             start = c->rrobin_index;
 874         else if (start == c->rrobin_index)
 875             return;
 876
 877         if (pa_memblockq_peek(r->memblockq,  &chunk) >= 0) {
 878             pa_memchunk schunk = chunk;
 879
 880             if (schunk.length > r->fragment_size)
 881                 schunk.length = r->fragment_size;
 882
 883             pa_pstream_send_memblock(c->pstream, r->index, 0, PA_SEEK_RELATIVE, &schunk);
 884
 885             pa_memblockq_drop(r->memblockq, schunk.length);
 886             pa_memblock_unref(schunk.memblock);
 887
 888             return;
 889         }
 890     }
 891 }
 892
 893 static void send_playback_stream_killed(playback_stream *p) {
 894     pa_tagstruct *t;
 895     playback_stream_assert_ref(p);
 896
 897     t = pa_tagstruct_new(NULL, 0);
 898     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_KILLED);
 899     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 900     pa_tagstruct_putu32(t, p->index);
 901     pa_pstream_send_tagstruct(p->connection->pstream, t);
 902 }
 903
 904 static void send_record_stream_killed(record_stream *r) {
 905     pa_tagstruct *t;
 906     record_stream_assert_ref(r);
 907
 908     t = pa_tagstruct_new(NULL, 0);
 909     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_KILLED);
 910     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 911     pa_tagstruct_putu32(t, r->index);
 912     pa_pstream_send_tagstruct(r->connection->pstream, t);
 913 }
 914
 915 /*** sink input callbacks ***/
 916
 917 /* Called from thread context */
 918 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
 919     pa_sink_input *i = PA_SINK_INPUT(o);
 920     playback_stream *s;
 921
 922     pa_sink_input_assert_ref(i);
 923     s = PLAYBACK_STREAM(i->userdata);
 924     playback_stream_assert_ref(s);
 925
 926     switch (code) {
 927
 928         case SINK_INPUT_MESSAGE_SEEK:
 929             pa_memblockq_seek(s->memblockq, offset, PA_PTR_TO_UINT(userdata));
 930             request_bytes(s);
 931             return 0;
 932
 933         case SINK_INPUT_MESSAGE_POST_DATA: {
 934             pa_assert(chunk);
 935
 936 /*             pa_log("sink input post: %u", chunk->length); */
 937
 938             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
 939
 940                 pa_log_warn("Failed to push data into queue");
 941                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_OVERFLOW, NULL, 0, NULL, NULL);
 942                 pa_memblockq_seek(s->memblockq, chunk->length, PA_SEEK_RELATIVE);
 943             }
 944
 945             request_bytes(s);
 946
 947             s->underrun = FALSE;
 948             return 0;
 949         }
 950
 951         case SINK_INPUT_MESSAGE_DRAIN: {
 952
 953             pa_memblockq_prebuf_disable(s->memblockq);
 954
 955             if (!pa_memblockq_is_readable(s->memblockq))
 956                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_DRAIN_ACK, userdata, 0, NULL, NULL);
 957             else {
 958                 s->drain_tag = PA_PTR_TO_UINT(userdata);
 959                 s->drain_request = TRUE;
 960             }
 961             request_bytes(s);
 962
 963             return 0;
 964         }
 965
 966         case SINK_INPUT_MESSAGE_FLUSH:
 967         case SINK_INPUT_MESSAGE_PREBUF_FORCE:
 968         case SINK_INPUT_MESSAGE_TRIGGER: {
 969
 970             pa_sink_input *isync;
 971             void (*func)(pa_memblockq *bq);
 972
 973             switch  (code) {
 974                 case SINK_INPUT_MESSAGE_FLUSH:
 975                     func = pa_memblockq_flush;
 976                     break;
 977
 978                 case SINK_INPUT_MESSAGE_PREBUF_FORCE:
 979                     func = pa_memblockq_prebuf_force;
 980                     break;
 981
 982                 case SINK_INPUT_MESSAGE_TRIGGER:
 983                     func = pa_memblockq_prebuf_disable;
 984                     break;
 985
 986                 default:
 987                     pa_assert_not_reached();
 988             }
 989
 990             func(s->memblockq);
 991             s->underrun = FALSE;
 992             request_bytes(s);
 993
 994             /* Do the same for all other members in the sync group */
 995             for (isync = i->sync_prev; isync; isync = isync->sync_prev) {
 996                 playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
 997                 func(ssync->memblockq);
 998                 ssync->underrun = FALSE;
 999                 request_bytes(ssync);
1000             }
1001
1002             for (isync = i->sync_next; isync; isync = isync->sync_next) {
1003                 playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
1004                 func(ssync->memblockq);
1005                 ssync->underrun = FALSE;
1006                 request_bytes(ssync);
1007             }
1008
1009             return 0;
1010         }
1011
1012         case SINK_INPUT_MESSAGE_UPDATE_LATENCY:
1013
1014             s->read_index = pa_memblockq_get_read_index(s->memblockq);
1015             s->write_index = pa_memblockq_get_write_index(s->memblockq);
1016             s->render_memblockq_length = pa_memblockq_get_length(s->sink_input->thread_info.render_memblockq);
1017             return 0;
1018
1019         case PA_SINK_INPUT_MESSAGE_SET_STATE:
1020
1021             pa_memblockq_prebuf_force(s->memblockq);
1022             request_bytes(s);
1023             break;
1024
1025         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
1026             pa_usec_t *r = userdata;
1027
1028             *r = pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
1029
1030             /* Fall through, the default handler will add in the extra
1031              * latency added by the resampler */
1032             break;
1033         }
1034     }
1035
1036     return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
1037 }
1038
1039 /* Called from thread context */
1040 static int sink_input_pop_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
1041     playback_stream *s;
1042
1043     pa_sink_input_assert_ref(i);
1044     s = PLAYBACK_STREAM(i->userdata);
1045     playback_stream_assert_ref(s);
1046     pa_assert(chunk);
1047
1048     if (pa_memblockq_peek(s->memblockq, chunk) < 0) {
1049
1050         if (s->drain_request && pa_sink_input_safe_to_remove(i)) {
1051             s->drain_request = FALSE;
1052             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_DRAIN_ACK, PA_UINT_TO_PTR(s->drain_tag), 0, NULL, NULL);
1053         } else if (!s->underrun) {
1054             s->underrun = TRUE;
1055             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_UNDERFLOW, NULL, 0, NULL, NULL);
1056         }
1057
1058         return -1;
1059     }
1060
1061     pa_memblockq_drop(s->memblockq, chunk->length);
1062     request_bytes(s);
1063
1064     return 0;
1065 }
1066
1067 /* Called from main context */
1068 static void sink_input_kill_cb(pa_sink_input *i) {
1069     playback_stream *s;
1070
1071     pa_sink_input_assert_ref(i);
1072     s = PLAYBACK_STREAM(i->userdata);
1073     playback_stream_assert_ref(s);
1074
1075     send_playback_stream_killed(s);
1076     playback_stream_unlink(s);
1077 }
1078
1079 /* Called from main context */
1080 static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspend) {
1081     playback_stream *s;
1082     pa_tagstruct *t;
1083
1084     pa_sink_input_assert_ref(i);
1085     s = PLAYBACK_STREAM(i->userdata);
1086     playback_stream_assert_ref(s);
1087
1088     if (s->connection->version < 12)
1089       return;
1090
1091     t = pa_tagstruct_new(NULL, 0);
1092     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_SUSPENDED);
1093     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1094     pa_tagstruct_putu32(t, s->index);
1095     pa_tagstruct_put_boolean(t, suspend);
1096     pa_pstream_send_tagstruct(s->connection->pstream, t);
1097 }
1098
1099 /* Called from main context */
1100 static void sink_input_moved_cb(pa_sink_input *i) {
1101     playback_stream *s;
1102     pa_tagstruct *t;
1103
1104     pa_sink_input_assert_ref(i);
1105     s = PLAYBACK_STREAM(i->userdata);
1106     playback_stream_assert_ref(s);
1107
1108     if (s->connection->version < 12)
1109       return;
1110
1111     t = pa_tagstruct_new(NULL, 0);
1112     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_MOVED);
1113     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1114     pa_tagstruct_putu32(t, s->index);
1115     pa_tagstruct_putu32(t, i->sink->index);
1116     pa_tagstruct_puts(t, i->sink->name);
1117     pa_tagstruct_put_boolean(t, pa_sink_get_state(i->sink) == PA_SINK_SUSPENDED);
1118     pa_pstream_send_tagstruct(s->connection->pstream, t);
1119 }
1120
1121 /*** source_output callbacks ***/
1122
1123 /* Called from thread context */
1124 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
1125     record_stream *s;
1126
1127     pa_source_output_assert_ref(o);
1128     s = RECORD_STREAM(o->userdata);
1129     record_stream_assert_ref(s);
1130     pa_assert(chunk);
1131
1132     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), RECORD_STREAM_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
1133 }
1134
1135 static void source_output_kill_cb(pa_source_output *o) {
1136     record_stream *s;
1137
1138     pa_source_output_assert_ref(o);
1139     s = RECORD_STREAM(o->userdata);
1140     record_stream_assert_ref(s);
1141
1142     send_record_stream_killed(s);
1143     record_stream_unlink(s);
1144 }
1145
1146 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
1147     record_stream *s;
1148
1149     pa_source_output_assert_ref(o);
1150     s = RECORD_STREAM(o->userdata);
1151     record_stream_assert_ref(s);
1152
1153     /*pa_log("get_latency: %u", pa_memblockq_get_length(s->memblockq));*/
1154
1155     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &o->sample_spec);
1156 }
1157
1158 /* Called from main context */
1159 static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend) {
1160     record_stream *s;
1161     pa_tagstruct *t;
1162
1163     pa_source_output_assert_ref(o);
1164     s = RECORD_STREAM(o->userdata);
1165     record_stream_assert_ref(s);
1166
1167     if (s->connection->version < 12)
1168       return;
1169
1170     t = pa_tagstruct_new(NULL, 0);
1171     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_SUSPENDED);
1172     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1173     pa_tagstruct_putu32(t, s->index);
1174     pa_tagstruct_put_boolean(t, suspend);
1175     pa_pstream_send_tagstruct(s->connection->pstream, t);
1176 }
1177
1178 /* Called from main context */
1179 static void source_output_moved_cb(pa_source_output *o) {
1180     record_stream *s;
1181     pa_tagstruct *t;
1182
1183     pa_source_output_assert_ref(o);
1184     s = RECORD_STREAM(o->userdata);
1185     record_stream_assert_ref(s);
1186
1187     if (s->connection->version < 12)
1188       return;
1189
1190     t = pa_tagstruct_new(NULL, 0);
1191     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_MOVED);
1192     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1193     pa_tagstruct_putu32(t, s->index);
1194     pa_tagstruct_putu32(t, o->source->index);
1195     pa_tagstruct_puts(t, o->source->name);
1196     pa_tagstruct_put_boolean(t, pa_source_get_state(o->source) == PA_SOURCE_SUSPENDED);
1197     pa_pstream_send_tagstruct(s->connection->pstream, t);
1198 }
1199
1200 /*** pdispatch callbacks ***/
1201
1202 static void protocol_error(connection *c) {
1203     pa_log("protocol error, kicking client");
1204     connection_unlink(c);
1205 }
1206
1207 #define CHECK_VALIDITY(pstream, expression, tag, error) do { \
1208 if (!(expression)) { \
1209     pa_pstream_send_error((pstream), (tag), (error)); \
1210     return; \
1211 } \
1212 } while(0);
1213
1214 static pa_tagstruct *reply_new(uint32_t tag) {
1215     pa_tagstruct *reply;
1216
1217     reply = pa_tagstruct_new(NULL, 0);
1218     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1219     pa_tagstruct_putu32(reply, tag);
1220     return reply;
1221 }
1222
1223 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) {
1224     connection *c = CONNECTION(userdata);
1225     playback_stream *s;
1226     uint32_t maxlength, tlength, prebuf, minreq, sink_index, syncid, missing;
1227     const char *name = NULL, *sink_name;
1228     pa_sample_spec ss;
1229     pa_channel_map map;
1230     pa_tagstruct *reply;
1231     pa_sink *sink = NULL;
1232     pa_cvolume volume;
1233     int corked;
1234     int no_remap = 0, no_remix = 0, fix_format = 0, fix_rate = 0, fix_channels = 0, no_move = 0, variable_rate = 0;
1235     pa_sink_input_flags_t flags = 0;
1236     pa_proplist *p;
1237
1238     connection_assert_ref(c);
1239     pa_assert(t);
1240
1241     if ((c->version < 13 && (pa_tagstruct_gets(t, &name) < 0 || !name)) ||
1242         pa_tagstruct_get(
1243                 t,
1244                 PA_TAG_SAMPLE_SPEC, &ss,
1245                 PA_TAG_CHANNEL_MAP, &map,
1246                 PA_TAG_U32, &sink_index,
1247                 PA_TAG_STRING, &sink_name,
1248                 PA_TAG_U32, &maxlength,
1249                 PA_TAG_BOOLEAN, &corked,
1250                 PA_TAG_U32, &tlength,
1251                 PA_TAG_U32, &prebuf,
1252                 PA_TAG_U32, &minreq,
1253                 PA_TAG_U32, &syncid,
1254                 PA_TAG_CVOLUME, &volume,
1255                 PA_TAG_INVALID) < 0) {
1256
1257         protocol_error(c);
1258         return;
1259     }
1260
1261     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1262     CHECK_VALIDITY(c->pstream, sink_index != PA_INVALID_INDEX || !sink_name || (*sink_name && pa_utf8_valid(sink_name)), tag, PA_ERR_INVALID);
1263     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1264     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1265     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
1266     CHECK_VALIDITY(c->pstream, map.channels == ss.channels && volume.channels == ss.channels, tag, PA_ERR_INVALID);
1267     CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
1268     CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
1269
1270     p = pa_proplist_new();
1271
1272     if (name)
1273         pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
1274
1275     if (c->version >= 12)  {
1276         /* Since 0.9.8 the user can ask for a couple of additional flags */
1277
1278         if (pa_tagstruct_get_boolean(t, &no_remap) < 0 ||
1279             pa_tagstruct_get_boolean(t, &no_remix) < 0 ||
1280             pa_tagstruct_get_boolean(t, &fix_format) < 0 ||
1281             pa_tagstruct_get_boolean(t, &fix_rate) < 0 ||
1282             pa_tagstruct_get_boolean(t, &fix_channels) < 0 ||
1283             pa_tagstruct_get_boolean(t, &no_move) < 0 ||
1284             pa_tagstruct_get_boolean(t, &variable_rate) < 0) {
1285
1286             protocol_error(c);
1287             pa_proplist_free(p);
1288             return;
1289         }
1290     }
1291
1292     if (c->version >= 13) {
1293
1294         if (pa_tagstruct_get_proplist(t, p) < 0) {
1295             protocol_error(c);
1296             pa_proplist_free(p);
1297             return;
1298         }
1299     }
1300
1301     if (!pa_tagstruct_eof(t)) {
1302         protocol_error(c);
1303         pa_proplist_free(p);
1304         return;
1305     }
1306
1307     if (sink_index != PA_INVALID_INDEX) {
1308
1309         if (!(sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index))) {
1310             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1311             pa_proplist_free(p);
1312             return;
1313         }
1314
1315     } else if (sink_name) {
1316
1317         if (!(sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1))) {
1318             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1319             pa_proplist_free(p);
1320             return;
1321         }
1322     }
1323
1324     flags =
1325         (corked ?  PA_SINK_INPUT_START_CORKED : 0) |
1326         (no_remap ?  PA_SINK_INPUT_NO_REMAP : 0) |
1327         (no_remix ?  PA_SINK_INPUT_NO_REMIX : 0) |
1328         (fix_format ?  PA_SINK_INPUT_FIX_FORMAT : 0) |
1329         (fix_rate ?  PA_SINK_INPUT_FIX_RATE : 0) |
1330         (fix_channels ?  PA_SINK_INPUT_FIX_CHANNELS : 0) |
1331         (no_move ?  PA_SINK_INPUT_DONT_MOVE : 0) |
1332         (variable_rate ?  PA_SINK_INPUT_VARIABLE_RATE : 0);
1333
1334     s = playback_stream_new(c, sink, &ss, &map, name, &maxlength, &tlength, &prebuf, &minreq, &volume, syncid, &missing, flags, p);
1335     pa_proplist_free(p);
1336
1337     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
1338
1339     reply = reply_new(tag);
1340     pa_tagstruct_putu32(reply, s->index);
1341     pa_assert(s->sink_input);
1342     pa_tagstruct_putu32(reply, s->sink_input->index);
1343     pa_tagstruct_putu32(reply, missing);
1344
1345 /*     pa_log("initial request is %u", missing); */
1346
1347     if (c->version >= 9) {
1348         /* Since 0.9.0 we support sending the buffer metrics back to the client */
1349
1350         pa_tagstruct_putu32(reply, (uint32_t) maxlength);
1351         pa_tagstruct_putu32(reply, (uint32_t) tlength);
1352         pa_tagstruct_putu32(reply, (uint32_t) prebuf);
1353         pa_tagstruct_putu32(reply, (uint32_t) minreq);
1354     }
1355
1356     if (c->version >= 12) {
1357         /* Since 0.9.8 we support sending the chosen sample
1358          * spec/channel map/device/suspend status back to the
1359          * client */
1360
1361         pa_tagstruct_put_sample_spec(reply, &ss);
1362         pa_tagstruct_put_channel_map(reply, &map);
1363
1364         pa_tagstruct_putu32(reply, s->sink_input->sink->index);
1365         pa_tagstruct_puts(reply, s->sink_input->sink->name);
1366
1367         pa_tagstruct_put_boolean(reply, pa_sink_get_state(s->sink_input->sink) == PA_SINK_SUSPENDED);
1368     }
1369
1370     pa_pstream_send_tagstruct(c->pstream, reply);
1371 }
1372
1373 static void command_delete_stream(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1374     connection *c = CONNECTION(userdata);
1375     uint32_t channel;
1376
1377     connection_assert_ref(c);
1378     pa_assert(t);
1379
1380     if (pa_tagstruct_getu32(t, &channel) < 0 ||
1381         !pa_tagstruct_eof(t)) {
1382         protocol_error(c);
1383         return;
1384     }
1385
1386     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1387
1388     switch (command) {
1389
1390         case PA_COMMAND_DELETE_PLAYBACK_STREAM: {
1391             playback_stream *s;
1392             if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !playback_stream_isinstance(s)) {
1393                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
1394                 return;
1395             }
1396
1397             playback_stream_unlink(s);
1398             break;
1399         }
1400
1401         case PA_COMMAND_DELETE_RECORD_STREAM: {
1402             record_stream *s;
1403             if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
1404                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
1405                 return;
1406             }
1407
1408             record_stream_unlink(s);
1409             break;
1410         }
1411
1412         case PA_COMMAND_DELETE_UPLOAD_STREAM: {
1413             upload_stream *s;
1414
1415             if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !upload_stream_isinstance(s)) {
1416                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
1417                 return;
1418             }
1419
1420             upload_stream_unlink(s);
1421             break;
1422         }
1423
1424         default:
1425             pa_assert_not_reached();
1426     }
1427
1428     pa_pstream_send_simple_ack(c->pstream, tag);
1429 }
1430
1431 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) {
1432     connection *c = CONNECTION(userdata);
1433     record_stream *s;
1434     uint32_t maxlength, fragment_size;
1435     uint32_t source_index;
1436     const char *name, *source_name;
1437     pa_sample_spec ss;
1438     pa_channel_map map;
1439     pa_tagstruct *reply;
1440     pa_source *source = NULL;
1441     int corked;
1442     int no_remap = 0, no_remix = 0, fix_format = 0, fix_rate = 0, fix_channels = 0, no_move = 0, variable_rate = 0;
1443     pa_source_output_flags_t flags = 0;
1444     pa_proplist *p;
1445
1446     connection_assert_ref(c);
1447     pa_assert(t);
1448
1449     if ((c->version < 13 && (pa_tagstruct_gets(t, &name) < 0 || !name)) ||
1450         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1451         pa_tagstruct_get_channel_map(t, &map) < 0 ||
1452         pa_tagstruct_getu32(t, &source_index) < 0 ||
1453         pa_tagstruct_gets(t, &source_name) < 0 ||
1454         pa_tagstruct_getu32(t, &maxlength) < 0 ||
1455         pa_tagstruct_get_boolean(t, &corked) < 0 ||
1456         pa_tagstruct_getu32(t, &fragment_size) < 0) {
1457         protocol_error(c);
1458         return;
1459     }
1460
1461     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1462     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1463     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1464     CHECK_VALIDITY(c->pstream, source_index != PA_INVALID_INDEX || !source_name || (*source_name && pa_utf8_valid(source_name)), tag, PA_ERR_INVALID);
1465     CHECK_VALIDITY(c->pstream, map.channels == ss.channels, tag, PA_ERR_INVALID);
1466     CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
1467     CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
1468
1469     p = pa_proplist_new();
1470
1471     if (name)
1472         pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
1473
1474     if (c->version >= 12)  {
1475         /* Since 0.9.8 the user can ask for a couple of additional flags */
1476
1477         if (pa_tagstruct_get_boolean(t, &no_remap) < 0 ||
1478             pa_tagstruct_get_boolean(t, &no_remix) < 0 ||
1479             pa_tagstruct_get_boolean(t, &fix_format) < 0 ||
1480             pa_tagstruct_get_boolean(t, &fix_rate) < 0 ||
1481             pa_tagstruct_get_boolean(t, &fix_channels) < 0 ||
1482             pa_tagstruct_get_boolean(t, &no_move) < 0 ||
1483             pa_tagstruct_get_boolean(t, &variable_rate) < 0) {
1484
1485             protocol_error(c);
1486             pa_proplist_free(p);
1487             return;
1488         }
1489     }
1490
1491     if (c->version >= 13) {
1492
1493         if (pa_tagstruct_get_proplist(t, p) < 0) {
1494             protocol_error(c);
1495             pa_proplist_free(p);
1496             return;
1497         }
1498     }
1499
1500     if (!pa_tagstruct_eof(t)) {
1501         protocol_error(c);
1502         pa_proplist_free(p);
1503         return;
1504     }
1505
1506     if (source_index != PA_INVALID_INDEX) {
1507
1508         if (!(source = pa_idxset_get_by_index(c->protocol->core->sources, source_index))) {
1509             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1510             pa_proplist_free(p);
1511             return;
1512         }
1513
1514     } else if (source_name) {
1515
1516         if (!(source = pa_namereg_get(c->protocol->core, source_name, PA_NAMEREG_SOURCE, 1))) {
1517             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1518             pa_proplist_free(p);
1519             return;
1520         }
1521     }
1522
1523     flags =
1524         (corked ?  PA_SOURCE_OUTPUT_START_CORKED : 0) |
1525         (no_remap ?  PA_SOURCE_OUTPUT_NO_REMAP : 0) |
1526         (no_remix ?  PA_SOURCE_OUTPUT_NO_REMIX : 0) |
1527         (fix_format ?  PA_SOURCE_OUTPUT_FIX_FORMAT : 0) |
1528         (fix_rate ?  PA_SOURCE_OUTPUT_FIX_RATE : 0) |
1529         (fix_channels ?  PA_SOURCE_OUTPUT_FIX_CHANNELS : 0) |
1530         (no_move ?  PA_SOURCE_OUTPUT_DONT_MOVE : 0) |
1531         (variable_rate ?  PA_SOURCE_OUTPUT_VARIABLE_RATE : 0);
1532
1533     s = record_stream_new(c, source, &ss, &map, name, &maxlength, fragment_size, flags, p);
1534     pa_proplist_free(p);
1535
1536     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
1537
1538     reply = reply_new(tag);
1539     pa_tagstruct_putu32(reply, s->index);
1540     pa_assert(s->source_output);
1541     pa_tagstruct_putu32(reply, s->source_output->index);
1542
1543     if (c->version >= 9) {
1544         /* Since 0.9 we support sending the buffer metrics back to the client */
1545
1546         pa_tagstruct_putu32(reply, (uint32_t) maxlength);
1547         pa_tagstruct_putu32(reply, (uint32_t) s->fragment_size);
1548     }
1549
1550     if (c->version >= 12) {
1551         /* Since 0.9.8 we support sending the chosen sample
1552          * spec/channel map/device/suspend status back to the
1553          * client */
1554
1555         pa_tagstruct_put_sample_spec(reply, &ss);
1556         pa_tagstruct_put_channel_map(reply, &map);
1557
1558         pa_tagstruct_putu32(reply, s->source_output->source->index);
1559         pa_tagstruct_puts(reply, s->source_output->source->name);
1560
1561         pa_tagstruct_put_boolean(reply, pa_source_get_state(s->source_output->source) == PA_SOURCE_SUSPENDED);
1562     }
1563
1564     pa_pstream_send_tagstruct(c->pstream, reply);
1565 }
1566
1567 static void command_exit(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1568     connection *c = CONNECTION(userdata);
1569
1570     connection_assert_ref(c);
1571     pa_assert(t);
1572
1573     if (!pa_tagstruct_eof(t)) {
1574         protocol_error(c);
1575         return;
1576     }
1577
1578     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1579
1580     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
1581     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
1582 }
1583
1584 static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1585     connection *c = CONNECTION(userdata);
1586     const void*cookie;
1587     pa_tagstruct *reply;
1588     char tmp[16];
1589
1590     connection_assert_ref(c);
1591     pa_assert(t);
1592
1593     if (pa_tagstruct_getu32(t, &c->version) < 0 ||
1594         pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
1595         !pa_tagstruct_eof(t)) {
1596         protocol_error(c);
1597         return;
1598     }
1599
1600     /* Minimum supported version */
1601     if (c->version < 8) {
1602         pa_pstream_send_error(c->pstream, tag, PA_ERR_VERSION);
1603         return;
1604     }
1605
1606     pa_snprintf(tmp, sizeof(tmp), "%u", c->version);
1607     pa_proplist_sets(c->client->proplist, "native-protocol.version", tmp);
1608
1609     if (!c->authorized) {
1610         int success = 0;
1611
1612 #ifdef HAVE_CREDS
1613         const pa_creds *creds;
1614
1615         if ((creds = pa_pdispatch_creds(pd))) {
1616             if (creds->uid == getuid())
1617                 success = 1;
1618             else if (c->protocol->auth_group) {
1619                 int r;
1620                 gid_t gid;
1621
1622                 if ((gid = pa_get_gid_of_group(c->protocol->auth_group)) == (gid_t) -1)
1623                     pa_log_warn("failed to get GID of group '%s'", c->protocol->auth_group);
1624                 else if (gid == creds->gid)
1625                     success = 1;
1626
1627                 if (!success) {
1628                     if ((r = pa_uid_in_group(creds->uid, c->protocol->auth_group)) < 0)
1629                         pa_log_warn("failed to check group membership.");
1630                     else if (r > 0)
1631                         success = 1;
1632                 }
1633             }
1634
1635             pa_log_info("Got credentials: uid=%lu gid=%lu success=%i",
1636                         (unsigned long) creds->uid,
1637                         (unsigned long) creds->gid,
1638                         success);
1639
1640             if (c->version >= 10 &&
1641                 pa_mempool_is_shared(c->protocol->core->mempool) &&
1642                 creds->uid == getuid()) {
1643
1644                 pa_pstream_use_shm(c->pstream, 1);
1645                 pa_log_info("Enabled SHM for new connection");
1646             }
1647
1648         }
1649 #endif
1650
1651         if (!success && memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) == 0)
1652             success = 1;
1653
1654         if (!success) {
1655             pa_log_warn("Denied access to client with invalid authorization data.");
1656             pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS);
1657             return;
1658         }
1659
1660         c->authorized = TRUE;
1661         if (c->auth_timeout_event) {
1662             c->protocol->core->mainloop->time_free(c->auth_timeout_event);
1663             c->auth_timeout_event = NULL;
1664         }
1665     }
1666
1667     reply = reply_new(tag);
1668     pa_tagstruct_putu32(reply, PA_PROTOCOL_VERSION);
1669
1670 #ifdef HAVE_CREDS
1671 {
1672     /* SHM support is only enabled after both sides made sure they are the same user. */
1673
1674     pa_creds ucred;
1675
1676     ucred.uid = getuid();
1677     ucred.gid = getgid();
1678
1679     pa_pstream_send_tagstruct_with_creds(c->pstream, reply, &ucred);
1680 }
1681 #else
1682     pa_pstream_send_tagstruct(c->pstream, reply);
1683 #endif
1684 }
1685
1686 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) {
1687     connection *c = CONNECTION(userdata);
1688     const char *name = NULL;
1689     pa_proplist *p;
1690     pa_tagstruct *reply;
1691
1692     connection_assert_ref(c);
1693     pa_assert(t);
1694
1695     p = pa_proplist_new();
1696
1697     if ((c->version < 13 && pa_tagstruct_gets(t, &name) < 0) ||
1698         (c->version >= 13 && pa_tagstruct_get_proplist(t, p) < 0) ||
1699         !pa_tagstruct_eof(t)) {
1700
1701         protocol_error(c);
1702         pa_proplist_free(p);
1703         return;
1704     }
1705
1706     if (name)
1707         if (pa_proplist_sets(p, PA_PROP_APPLICATION_NAME, name) < 0) {
1708             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
1709             pa_proplist_free(p);
1710             return;
1711         }
1712
1713     pa_proplist_update(c->client->proplist, PA_UPDATE_REPLACE, p);
1714     pa_proplist_free(p);
1715
1716     pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_CLIENT|PA_SUBSCRIPTION_EVENT_CHANGE, c->client->index);
1717
1718     reply = reply_new(tag);
1719
1720     if (c->version >= 13)
1721         pa_tagstruct_putu32(reply, c->client->index);
1722
1723     pa_pstream_send_tagstruct(c->pstream, reply);
1724 }
1725
1726 static void command_lookup(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1727     connection *c = CONNECTION(userdata);
1728     const char *name;
1729     uint32_t idx = PA_IDXSET_INVALID;
1730
1731     connection_assert_ref(c);
1732     pa_assert(t);
1733
1734     if (pa_tagstruct_gets(t, &name) < 0 ||
1735         !pa_tagstruct_eof(t)) {
1736         protocol_error(c);
1737         return;
1738     }
1739
1740     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1741     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1742
1743     if (command == PA_COMMAND_LOOKUP_SINK) {
1744         pa_sink *sink;
1745         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1)))
1746             idx = sink->index;
1747     } else {
1748         pa_source *source;
1749         pa_assert(command == PA_COMMAND_LOOKUP_SOURCE);
1750         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1)))
1751             idx = source->index;
1752     }
1753
1754     if (idx == PA_IDXSET_INVALID)
1755         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1756     else {
1757         pa_tagstruct *reply;
1758         reply = reply_new(tag);
1759         pa_tagstruct_putu32(reply, idx);
1760         pa_pstream_send_tagstruct(c->pstream, reply);
1761     }
1762 }
1763
1764 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) {
1765     connection *c = CONNECTION(userdata);
1766     uint32_t idx;
1767     playback_stream *s;
1768
1769     connection_assert_ref(c);
1770     pa_assert(t);
1771
1772     if (pa_tagstruct_getu32(t, &idx) < 0 ||
1773         !pa_tagstruct_eof(t)) {
1774         protocol_error(c);
1775         return;
1776     }
1777
1778     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1779     s = pa_idxset_get_by_index(c->output_streams, idx);
1780     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1781     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
1782
1783     pa_asyncmsgq_post(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_DRAIN, PA_UINT_TO_PTR(tag), 0, NULL, NULL);
1784 }
1785
1786 static void command_stat(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1787     connection *c = CONNECTION(userdata);
1788     pa_tagstruct *reply;
1789     const pa_mempool_stat *stat;
1790
1791     connection_assert_ref(c);
1792     pa_assert(t);
1793
1794     if (!pa_tagstruct_eof(t)) {
1795         protocol_error(c);
1796         return;
1797     }
1798
1799     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1800
1801     stat = pa_mempool_get_stat(c->protocol->core->mempool);
1802
1803     reply = reply_new(tag);
1804     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->n_allocated));
1805     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->allocated_size));
1806     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->n_accumulated));
1807     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->accumulated_size));
1808     pa_tagstruct_putu32(reply, pa_scache_total_size(c->protocol->core));
1809     pa_pstream_send_tagstruct(c->pstream, reply);
1810 }
1811
1812 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) {
1813     connection *c = CONNECTION(userdata);
1814     pa_tagstruct *reply;
1815     playback_stream *s;
1816     struct timeval tv, now;
1817     uint32_t idx;
1818     pa_usec_t latency;
1819
1820     connection_assert_ref(c);
1821     pa_assert(t);
1822
1823     if (pa_tagstruct_getu32(t, &idx) < 0 ||
1824         pa_tagstruct_get_timeval(t, &tv) < 0 ||
1825         !pa_tagstruct_eof(t)) {
1826         protocol_error(c);
1827         return;
1828     }
1829
1830     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1831     s = pa_idxset_get_by_index(c->output_streams, idx);
1832     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1833     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
1834     CHECK_VALIDITY(c->pstream, pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_UPDATE_LATENCY, s, 0, NULL) == 0, tag, PA_ERR_NOENTITY)
1835
1836     reply = reply_new(tag);
1837
1838     latency = pa_sink_get_latency(s->sink_input->sink);
1839     latency += pa_bytes_to_usec(s->render_memblockq_length, &s->sink_input->sample_spec);
1840
1841     pa_tagstruct_put_usec(reply, latency);
1842
1843     pa_tagstruct_put_usec(reply, 0);
1844     pa_tagstruct_put_boolean(reply, pa_sink_input_get_state(s->sink_input) == PA_SINK_INPUT_RUNNING);
1845     pa_tagstruct_put_timeval(reply, &tv);
1846     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
1847     pa_tagstruct_puts64(reply, s->write_index);
1848     pa_tagstruct_puts64(reply, s->read_index);
1849     pa_pstream_send_tagstruct(c->pstream, reply);
1850 }
1851
1852 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) {
1853     connection *c = CONNECTION(userdata);
1854     pa_tagstruct *reply;
1855     record_stream *s;
1856     struct timeval tv, now;
1857     uint32_t idx;
1858
1859     connection_assert_ref(c);
1860     pa_assert(t);
1861
1862     if (pa_tagstruct_getu32(t, &idx) < 0 ||
1863         pa_tagstruct_get_timeval(t, &tv) < 0 ||
1864         !pa_tagstruct_eof(t)) {
1865         protocol_error(c);
1866         return;
1867     }
1868
1869     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1870     s = pa_idxset_get_by_index(c->record_streams, idx);
1871     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1872
1873     reply = reply_new(tag);
1874     pa_tagstruct_put_usec(reply, s->source_output->source->monitor_of ? pa_sink_get_latency(s->source_output->source->monitor_of) : 0);
1875     pa_tagstruct_put_usec(reply, pa_source_get_latency(s->source_output->source));
1876     pa_tagstruct_put_boolean(reply, 0);
1877     pa_tagstruct_put_timeval(reply, &tv);
1878     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
1879     pa_tagstruct_puts64(reply, pa_memblockq_get_write_index(s->memblockq));
1880     pa_tagstruct_puts64(reply, pa_memblockq_get_read_index(s->memblockq));
1881     pa_pstream_send_tagstruct(c->pstream, reply);
1882 }
1883
1884 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) {
1885     connection *c = CONNECTION(userdata);
1886     upload_stream *s;
1887     uint32_t length;
1888     const char *name = NULL;
1889     pa_sample_spec ss;
1890     pa_channel_map map;
1891     pa_tagstruct *reply;
1892     pa_proplist *p;
1893
1894     connection_assert_ref(c);
1895     pa_assert(t);
1896
1897     if ((c->version < 13 && pa_tagstruct_gets(t, &name) < 0) ||
1898         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1899         pa_tagstruct_get_channel_map(t, &map) < 0 ||
1900         pa_tagstruct_getu32(t, &length) < 0) {
1901         protocol_error(c);
1902         return;
1903     }
1904
1905     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1906     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1907     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1908     CHECK_VALIDITY(c->pstream, map.channels == ss.channels, tag, PA_ERR_INVALID);
1909     CHECK_VALIDITY(c->pstream, (length % pa_frame_size(&ss)) == 0 && length > 0, tag, PA_ERR_INVALID);
1910     CHECK_VALIDITY(c->pstream, length <= PA_SCACHE_ENTRY_SIZE_MAX, tag, PA_ERR_TOOLARGE);
1911
1912     if (c->version < 13)
1913         CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1914
1915     p = pa_proplist_new();
1916
1917     if (c->version >= 13 && pa_tagstruct_get_proplist(t, p) < 0) {
1918         protocol_error(c);
1919         pa_proplist_free(p);
1920         return;
1921     }
1922
1923     if (c->version < 13)
1924         pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
1925
1926     s = upload_stream_new(c, &ss, &map, name, length, p);
1927     pa_proplist_free(p);
1928
1929     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
1930
1931     reply = reply_new(tag);
1932     pa_tagstruct_putu32(reply, s->index);
1933     pa_tagstruct_putu32(reply, length);
1934     pa_pstream_send_tagstruct(c->pstream, reply);
1935 }
1936
1937 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) {
1938     connection *c = CONNECTION(userdata);
1939     uint32_t channel;
1940     upload_stream *s;
1941     uint32_t idx;
1942
1943     connection_assert_ref(c);
1944     pa_assert(t);
1945
1946     if (pa_tagstruct_getu32(t, &channel) < 0 ||
1947         !pa_tagstruct_eof(t)) {
1948         protocol_error(c);
1949         return;
1950     }
1951
1952     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1953
1954     s = pa_idxset_get_by_index(c->output_streams, channel);
1955     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1956     CHECK_VALIDITY(c->pstream, upload_stream_isinstance(s), tag, PA_ERR_NOENTITY);
1957
1958     if (pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->channel_map, &s->memchunk, s->proplist, &idx) < 0)
1959         pa_pstream_send_error(c->pstream, tag, PA_ERR_INTERNAL);
1960     else
1961         pa_pstream_send_simple_ack(c->pstream, tag);
1962
1963     upload_stream_unlink(s);
1964 }
1965
1966 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) {
1967     connection *c = CONNECTION(userdata);
1968     uint32_t sink_index;
1969     pa_volume_t volume;
1970     pa_sink *sink;
1971     const char *name, *sink_name;
1972     uint32_t idx;
1973     pa_proplist *p;
1974     pa_tagstruct *reply;
1975
1976     connection_assert_ref(c);
1977     pa_assert(t);
1978
1979     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1980
1981     if (pa_tagstruct_getu32(t, &sink_index) < 0 ||
1982         pa_tagstruct_gets(t, &sink_name) < 0 ||
1983         pa_tagstruct_getu32(t, &volume) < 0 ||
1984         pa_tagstruct_gets(t, &name) < 0) {
1985         protocol_error(c);
1986         return;
1987     }
1988
1989     CHECK_VALIDITY(c->pstream, sink_index != PA_INVALID_INDEX || !sink_name || (*sink_name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
1990     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1991
1992     if (sink_index != PA_INVALID_INDEX)
1993         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
1994     else
1995         sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1);
1996
1997     CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
1998
1999     p = pa_proplist_new();
2000
2001     if ((c->version >= 13 && pa_tagstruct_get_proplist(t, p) < 0) ||
2002         !pa_tagstruct_eof(t)) {
2003         protocol_error(c);
2004         pa_proplist_free(p);
2005         return;
2006     }
2007
2008     if (pa_scache_play_item(c->protocol->core, name, sink, volume, p, &idx) < 0) {
2009         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2010         pa_proplist_free(p);
2011         return;
2012     }
2013
2014     pa_proplist_free(p);
2015
2016     reply = reply_new(tag);
2017
2018     if (c->version >= 13)
2019         pa_tagstruct_putu32(reply, idx);
2020
2021     pa_pstream_send_tagstruct(c->pstream, reply);
2022 }
2023
2024 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) {
2025     connection *c = CONNECTION(userdata);
2026     const char *name;
2027
2028     connection_assert_ref(c);
2029     pa_assert(t);
2030
2031     if (pa_tagstruct_gets(t, &name) < 0 ||
2032         !pa_tagstruct_eof(t)) {
2033         protocol_error(c);
2034         return;
2035     }
2036
2037     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2038     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
2039
2040     if (pa_scache_remove_item(c->protocol->core, name) < 0) {
2041         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2042         return;
2043     }
2044
2045     pa_pstream_send_simple_ack(c->pstream, tag);
2046 }
2047
2048 static void fixup_sample_spec(connection *c, pa_sample_spec *fixed, const pa_sample_spec *original) {
2049     pa_assert(c);
2050     pa_assert(fixed);
2051     pa_assert(original);
2052
2053     *fixed = *original;
2054
2055     if (c->version < 12) {
2056         /* Before protocol version 12 we didn't support S32 samples,
2057          * so we need to lie about this to the client */
2058
2059         if (fixed->format == PA_SAMPLE_S32LE)
2060             fixed->format = PA_SAMPLE_FLOAT32LE;
2061         if (fixed->format == PA_SAMPLE_S32BE)
2062             fixed->format = PA_SAMPLE_FLOAT32BE;
2063     }
2064 }
2065
2066 static void sink_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sink *sink) {
2067     pa_sample_spec fixed_ss;
2068
2069     pa_assert(t);
2070     pa_sink_assert_ref(sink);
2071
2072     fixup_sample_spec(c, &fixed_ss, &sink->sample_spec);
2073
2074     pa_tagstruct_put(
2075         t,
2076         PA_TAG_U32, sink->index,
2077         PA_TAG_STRING, sink->name,
2078         PA_TAG_STRING, pa_strnull(pa_proplist_gets(sink->proplist, PA_PROP_DEVICE_DESCRIPTION)),
2079         PA_TAG_SAMPLE_SPEC, &fixed_ss,
2080         PA_TAG_CHANNEL_MAP, &sink->channel_map,
2081         PA_TAG_U32, sink->module ? sink->module->index : PA_INVALID_INDEX,
2082         PA_TAG_CVOLUME, pa_sink_get_volume(sink),
2083         PA_TAG_BOOLEAN, pa_sink_get_mute(sink),
2084         PA_TAG_U32, sink->monitor_source ? sink->monitor_source->index : PA_INVALID_INDEX,
2085         PA_TAG_STRING, sink->monitor_source ? sink->monitor_source->name : NULL,
2086         PA_TAG_USEC, pa_sink_get_latency(sink),
2087         PA_TAG_STRING, sink->driver,
2088         PA_TAG_U32, sink->flags,
2089         PA_TAG_INVALID);
2090
2091     if (c->version >= 13)
2092         pa_tagstruct_put_proplist(t, sink->proplist);
2093 }
2094
2095 static void source_fill_tagstruct(connection *c, pa_tagstruct *t, pa_source *source) {
2096     pa_sample_spec fixed_ss;
2097
2098     pa_assert(t);
2099     pa_source_assert_ref(source);
2100
2101     fixup_sample_spec(c, &fixed_ss, &source->sample_spec);
2102
2103     pa_tagstruct_put(
2104         t,
2105         PA_TAG_U32, source->index,
2106         PA_TAG_STRING, source->name,
2107         PA_TAG_STRING, pa_strnull(pa_proplist_gets(source->proplist, PA_PROP_DEVICE_DESCRIPTION)),
2108         PA_TAG_SAMPLE_SPEC, &fixed_ss,
2109         PA_TAG_CHANNEL_MAP, &source->channel_map,
2110         PA_TAG_U32, source->module ? source->module->index : PA_INVALID_INDEX,
2111         PA_TAG_CVOLUME, pa_source_get_volume(source),
2112         PA_TAG_BOOLEAN, pa_source_get_mute(source),
2113         PA_TAG_U32, source->monitor_of ? source->monitor_of->index : PA_INVALID_INDEX,
2114         PA_TAG_STRING, source->monitor_of ? source->monitor_of->name : NULL,
2115         PA_TAG_USEC, pa_source_get_latency(source),
2116         PA_TAG_STRING, source->driver,
2117         PA_TAG_U32, source->flags,
2118         PA_TAG_INVALID);
2119
2120     if (c->version >= 13)
2121         pa_tagstruct_put_proplist(t, source->proplist);
2122 }
2123
2124 static void client_fill_tagstruct(connection *c, pa_tagstruct *t, pa_client *client) {
2125     pa_assert(t);
2126     pa_assert(client);
2127
2128     pa_tagstruct_putu32(t, client->index);
2129     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(client->proplist, PA_PROP_APPLICATION_NAME)));
2130     pa_tagstruct_putu32(t, client->module ? client->module->index : PA_INVALID_INDEX);
2131     pa_tagstruct_puts(t, client->driver);
2132
2133     if (c->version >= 13)
2134         pa_tagstruct_put_proplist(t, client->proplist);
2135
2136 }
2137
2138 static void module_fill_tagstruct(pa_tagstruct *t, pa_module *module) {
2139     pa_assert(t);
2140     pa_assert(module);
2141
2142     pa_tagstruct_putu32(t, module->index);
2143     pa_tagstruct_puts(t, module->name);
2144     pa_tagstruct_puts(t, module->argument);
2145     pa_tagstruct_putu32(t, module->n_used);
2146     pa_tagstruct_put_boolean(t, module->auto_unload);
2147 }
2148
2149 static void sink_input_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sink_input *s) {
2150     pa_sample_spec fixed_ss;
2151
2152     pa_assert(t);
2153     pa_sink_input_assert_ref(s);
2154
2155     fixup_sample_spec(c, &fixed_ss, &s->sample_spec);
2156
2157     pa_tagstruct_putu32(t, s->index);
2158     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME)));
2159     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
2160     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
2161     pa_tagstruct_putu32(t, s->sink->index);
2162     pa_tagstruct_put_sample_spec(t, &fixed_ss);
2163     pa_tagstruct_put_channel_map(t, &s->channel_map);
2164     pa_tagstruct_put_cvolume(t, &s->volume);
2165     pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s));
2166     pa_tagstruct_put_usec(t, pa_sink_get_latency(s->sink));
2167     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
2168     pa_tagstruct_puts(t, s->driver);
2169     if (c->version >= 11)
2170         pa_tagstruct_put_boolean(t, pa_sink_input_get_mute(s));
2171     if (c->version >= 13)
2172         pa_tagstruct_put_proplist(t, s->proplist);
2173 }
2174
2175 static void source_output_fill_tagstruct(connection *c, pa_tagstruct *t, pa_source_output *s) {
2176     pa_sample_spec fixed_ss;
2177
2178     pa_assert(t);
2179     pa_source_output_assert_ref(s);
2180
2181     fixup_sample_spec(c, &fixed_ss, &s->sample_spec);
2182
2183     pa_tagstruct_putu32(t, s->index);
2184     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME)));
2185     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
2186     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
2187     pa_tagstruct_putu32(t, s->source->index);
2188     pa_tagstruct_put_sample_spec(t, &fixed_ss);
2189     pa_tagstruct_put_channel_map(t, &s->channel_map);
2190     pa_tagstruct_put_usec(t, pa_source_output_get_latency(s));
2191     pa_tagstruct_put_usec(t, pa_source_get_latency(s->source));
2192     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_source_output_get_resample_method(s)));
2193     pa_tagstruct_puts(t, s->driver);
2194
2195     if (c->version >= 13)
2196         pa_tagstruct_put_proplist(t, s->proplist);
2197 }
2198
2199 static void scache_fill_tagstruct(connection *c, pa_tagstruct *t, pa_scache_entry *e) {
2200     pa_sample_spec fixed_ss;
2201
2202     pa_assert(t);
2203     pa_assert(e);
2204
2205     fixup_sample_spec(c, &fixed_ss, &e->sample_spec);
2206
2207     pa_tagstruct_putu32(t, e->index);
2208     pa_tagstruct_puts(t, e->name);
2209     pa_tagstruct_put_cvolume(t, &e->volume);
2210     pa_tagstruct_put_usec(t, pa_bytes_to_usec(e->memchunk.length, &e->sample_spec));
2211     pa_tagstruct_put_sample_spec(t, &fixed_ss);
2212     pa_tagstruct_put_channel_map(t, &e->channel_map);
2213     pa_tagstruct_putu32(t, e->memchunk.length);
2214     pa_tagstruct_put_boolean(t, e->lazy);
2215     pa_tagstruct_puts(t, e->filename);
2216
2217     if (c->version >= 13)
2218         pa_tagstruct_put_proplist(t, e->proplist);
2219 }
2220
2221 static void command_get_info(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2222     connection *c = CONNECTION(userdata);
2223     uint32_t idx;
2224     pa_sink *sink = NULL;
2225     pa_source *source = NULL;
2226     pa_client *client = NULL;
2227     pa_module *module = NULL;
2228     pa_sink_input *si = NULL;
2229     pa_source_output *so = NULL;
2230     pa_scache_entry *sce = NULL;
2231     const char *name;
2232     pa_tagstruct *reply;
2233
2234     connection_assert_ref(c);
2235     pa_assert(t);
2236
2237     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2238         (command != PA_COMMAND_GET_CLIENT_INFO &&
2239          command != PA_COMMAND_GET_MODULE_INFO &&
2240          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
2241          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
2242          pa_tagstruct_gets(t, &name) < 0) ||
2243         !pa_tagstruct_eof(t)) {
2244         protocol_error(c);
2245         return;
2246     }
2247
2248     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2249     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2250
2251     if (command == PA_COMMAND_GET_SINK_INFO) {
2252         if (idx != PA_INVALID_INDEX)
2253             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
2254         else
2255             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2256     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
2257         if (idx != PA_INVALID_INDEX)
2258             source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
2259         else
2260             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2261     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
2262         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
2263     else if (command == PA_COMMAND_GET_MODULE_INFO)
2264         module = pa_idxset_get_by_index(c->protocol->core->modules, idx);
2265     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
2266         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2267     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
2268         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
2269     else {
2270         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO);
2271         if (idx != PA_INVALID_INDEX)
2272             sce = pa_idxset_get_by_index(c->protocol->core->scache, idx);
2273         else
2274             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE, 0);
2275     }
2276
2277     if (!sink && !source && !client && !module && !si && !so && !sce) {
2278         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2279         return;
2280     }
2281
2282     reply = reply_new(tag);
2283     if (sink)
2284         sink_fill_tagstruct(c, reply, sink);
2285     else if (source)
2286         source_fill_tagstruct(c, reply, source);
2287     else if (client)
2288         client_fill_tagstruct(c, reply, client);
2289     else if (module)
2290         module_fill_tagstruct(reply, module);
2291     else if (si)
2292         sink_input_fill_tagstruct(c, reply, si);
2293     else if (so)
2294         source_output_fill_tagstruct(c, reply, so);
2295     else
2296         scache_fill_tagstruct(c, reply, sce);
2297     pa_pstream_send_tagstruct(c->pstream, reply);
2298 }
2299
2300 static void command_get_info_list(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2301     connection *c = CONNECTION(userdata);
2302     pa_idxset *i;
2303     uint32_t idx;
2304     void *p;
2305     pa_tagstruct *reply;
2306
2307     connection_assert_ref(c);
2308     pa_assert(t);
2309
2310     if (!pa_tagstruct_eof(t)) {
2311         protocol_error(c);
2312         return;
2313     }
2314
2315     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2316
2317     reply = reply_new(tag);
2318
2319     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
2320         i = c->protocol->core->sinks;
2321     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
2322         i = c->protocol->core->sources;
2323     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
2324         i = c->protocol->core->clients;
2325     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
2326         i = c->protocol->core->modules;
2327     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
2328         i = c->protocol->core->sink_inputs;
2329     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
2330         i = c->protocol->core->source_outputs;
2331     else {
2332         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
2333         i = c->protocol->core->scache;
2334     }
2335
2336     if (i) {
2337         for (p = pa_idxset_first(i, &idx); p; p = pa_idxset_next(i, &idx)) {
2338             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
2339                 sink_fill_tagstruct(c, reply, p);
2340             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
2341                 source_fill_tagstruct(c, reply, p);
2342             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
2343                 client_fill_tagstruct(c, reply, p);
2344             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
2345                 module_fill_tagstruct(reply, p);
2346             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
2347                 sink_input_fill_tagstruct(c, reply, p);
2348             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
2349                 source_output_fill_tagstruct(c, reply, p);
2350             else {
2351                 pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
2352                 scache_fill_tagstruct(c, reply, p);
2353             }
2354         }
2355     }
2356
2357     pa_pstream_send_tagstruct(c->pstream, reply);
2358 }
2359
2360 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) {
2361     connection *c = CONNECTION(userdata);
2362     pa_tagstruct *reply;
2363     char txt[256];
2364     const char *n;
2365     pa_sample_spec fixed_ss;
2366
2367     connection_assert_ref(c);
2368     pa_assert(t);
2369
2370     if (!pa_tagstruct_eof(t)) {
2371         protocol_error(c);
2372         return;
2373     }
2374
2375     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2376
2377     reply = reply_new(tag);
2378     pa_tagstruct_puts(reply, PACKAGE_NAME);
2379     pa_tagstruct_puts(reply, PACKAGE_VERSION);
2380     pa_tagstruct_puts(reply, pa_get_user_name(txt, sizeof(txt)));
2381     pa_tagstruct_puts(reply, pa_get_fqdn(txt, sizeof(txt)));
2382
2383     fixup_sample_spec(c, &fixed_ss, &c->protocol->core->default_sample_spec);
2384     pa_tagstruct_put_sample_spec(reply, &fixed_ss);
2385
2386     n = pa_namereg_get_default_sink_name(c->protocol->core);
2387     pa_tagstruct_puts(reply, n);
2388     n = pa_namereg_get_default_source_name(c->protocol->core);
2389     pa_tagstruct_puts(reply, n);
2390
2391     pa_tagstruct_putu32(reply, c->protocol->core->cookie);
2392
2393     pa_pstream_send_tagstruct(c->pstream, reply);
2394 }
2395
2396 static void subscription_cb(pa_core *core, pa_subscription_event_type_t e, uint32_t idx, void *userdata) {
2397     pa_tagstruct *t;
2398     connection *c = CONNECTION(userdata);
2399
2400     connection_assert_ref(c);
2401
2402     t = pa_tagstruct_new(NULL, 0);
2403     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
2404     pa_tagstruct_putu32(t, (uint32_t) -1);
2405     pa_tagstruct_putu32(t, e);
2406     pa_tagstruct_putu32(t, idx);
2407     pa_pstream_send_tagstruct(c->pstream, t);
2408 }
2409
2410 static void command_subscribe(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2411     connection *c = CONNECTION(userdata);
2412     pa_subscription_mask_t m;
2413
2414     connection_assert_ref(c);
2415     pa_assert(t);
2416
2417     if (pa_tagstruct_getu32(t, &m) < 0 ||
2418         !pa_tagstruct_eof(t)) {
2419         protocol_error(c);
2420         return;
2421     }
2422
2423     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2424     CHECK_VALIDITY(c->pstream, (m & ~PA_SUBSCRIPTION_MASK_ALL) == 0, tag, PA_ERR_INVALID);
2425
2426     if (c->subscription)
2427         pa_subscription_free(c->subscription);
2428
2429     if (m != 0) {
2430         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
2431         pa_assert(c->subscription);
2432     } else
2433         c->subscription = NULL;
2434
2435     pa_pstream_send_simple_ack(c->pstream, tag);
2436 }
2437
2438 static void command_set_volume(
2439         PA_GCC_UNUSED pa_pdispatch *pd,
2440         uint32_t command,
2441         uint32_t tag,
2442         pa_tagstruct *t,
2443         void *userdata) {
2444
2445     connection *c = CONNECTION(userdata);
2446     uint32_t idx;
2447     pa_cvolume volume;
2448     pa_sink *sink = NULL;
2449     pa_source *source = NULL;
2450     pa_sink_input *si = NULL;
2451     const char *name = NULL;
2452
2453     connection_assert_ref(c);
2454     pa_assert(t);
2455
2456     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2457         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
2458         (command == PA_COMMAND_SET_SOURCE_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
2459         pa_tagstruct_get_cvolume(t, &volume) ||
2460         !pa_tagstruct_eof(t)) {
2461         protocol_error(c);
2462         return;
2463     }
2464
2465     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2466     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2467     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
2468
2469     switch (command) {
2470
2471         case PA_COMMAND_SET_SINK_VOLUME:
2472             if (idx != PA_INVALID_INDEX)
2473                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
2474             else
2475                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2476             break;
2477
2478         case PA_COMMAND_SET_SOURCE_VOLUME:
2479             if (idx != PA_INVALID_INDEX)
2480                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
2481             else
2482                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2483             break;
2484
2485         case PA_COMMAND_SET_SINK_INPUT_VOLUME:
2486             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2487             break;
2488
2489         default:
2490             pa_assert_not_reached();
2491     }
2492
2493     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
2494
2495     if (sink)
2496         pa_sink_set_volume(sink, &volume);
2497     else if (source)
2498         pa_source_set_volume(source, &volume);
2499     else if (si)
2500         pa_sink_input_set_volume(si, &volume);
2501
2502     pa_pstream_send_simple_ack(c->pstream, tag);
2503 }
2504
2505 static void command_set_mute(
2506         PA_GCC_UNUSED pa_pdispatch *pd,
2507         uint32_t command,
2508         uint32_t tag,
2509         pa_tagstruct *t,
2510         void *userdata) {
2511
2512     connection *c = CONNECTION(userdata);
2513     uint32_t idx;
2514     int mute;
2515     pa_sink *sink = NULL;
2516     pa_source *source = NULL;
2517     pa_sink_input *si = NULL;
2518     const char *name = NULL;
2519
2520     connection_assert_ref(c);
2521     pa_assert(t);
2522
2523     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2524         (command == PA_COMMAND_SET_SINK_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
2525         (command == PA_COMMAND_SET_SOURCE_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
2526         pa_tagstruct_get_boolean(t, &mute) ||
2527         !pa_tagstruct_eof(t)) {
2528         protocol_error(c);
2529         return;
2530     }
2531
2532     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2533     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2534
2535     switch (command) {
2536
2537         case PA_COMMAND_SET_SINK_MUTE:
2538
2539             if (idx != PA_INVALID_INDEX)
2540                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
2541             else
2542                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2543
2544             break;
2545
2546         case PA_COMMAND_SET_SOURCE_MUTE:
2547             if (idx != PA_INVALID_INDEX)
2548                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
2549             else
2550                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2551
2552             break;
2553
2554         case PA_COMMAND_SET_SINK_INPUT_MUTE:
2555             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2556             break;
2557
2558         default:
2559             pa_assert_not_reached();
2560     }
2561
2562     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
2563
2564     if (sink)
2565         pa_sink_set_mute(sink, mute);
2566     else if (source)
2567         pa_source_set_mute(source, mute);
2568     else if (si)
2569         pa_sink_input_set_mute(si, mute);
2570
2571     pa_pstream_send_simple_ack(c->pstream, tag);
2572 }
2573
2574 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) {
2575     connection *c = CONNECTION(userdata);
2576     uint32_t idx;
2577     int b;
2578     playback_stream *s;
2579
2580     connection_assert_ref(c);
2581     pa_assert(t);
2582
2583     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2584         pa_tagstruct_get_boolean(t, &b) < 0 ||
2585         !pa_tagstruct_eof(t)) {
2586         protocol_error(c);
2587         return;
2588     }
2589
2590     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2591     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2592     s = pa_idxset_get_by_index(c->output_streams, idx);
2593     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2594     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2595
2596     pa_sink_input_cork(s->sink_input, b);
2597     pa_pstream_send_simple_ack(c->pstream, tag);
2598 }
2599
2600 static void command_trigger_or_flush_or_prebuf_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2601     connection *c = CONNECTION(userdata);
2602     uint32_t idx;
2603     playback_stream *s;
2604
2605     connection_assert_ref(c);
2606     pa_assert(t);
2607
2608     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2609         !pa_tagstruct_eof(t)) {
2610         protocol_error(c);
2611         return;
2612     }
2613
2614     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2615     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2616     s = pa_idxset_get_by_index(c->output_streams, idx);
2617     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2618     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2619
2620     switch (command) {
2621         case PA_COMMAND_FLUSH_PLAYBACK_STREAM:
2622             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_FLUSH, NULL, 0, NULL);
2623             break;
2624
2625         case PA_COMMAND_PREBUF_PLAYBACK_STREAM:
2626             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_PREBUF_FORCE, NULL, 0, NULL);
2627             break;
2628
2629         case PA_COMMAND_TRIGGER_PLAYBACK_STREAM:
2630             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_TRIGGER, NULL, 0, NULL);
2631             break;
2632
2633         default:
2634             pa_assert_not_reached();
2635     }
2636
2637     pa_pstream_send_simple_ack(c->pstream, tag);
2638 }
2639
2640 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) {
2641     connection *c = CONNECTION(userdata);
2642     uint32_t idx;
2643     record_stream *s;
2644     int b;
2645
2646     connection_assert_ref(c);
2647     pa_assert(t);
2648
2649     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2650         pa_tagstruct_get_boolean(t, &b) < 0 ||
2651         !pa_tagstruct_eof(t)) {
2652         protocol_error(c);
2653         return;
2654     }
2655
2656     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2657     s = pa_idxset_get_by_index(c->record_streams, idx);
2658     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2659
2660     pa_source_output_cork(s->source_output, b);
2661     pa_memblockq_prebuf_force(s->memblockq);
2662     pa_pstream_send_simple_ack(c->pstream, tag);
2663 }
2664
2665 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) {
2666     connection *c = CONNECTION(userdata);
2667     uint32_t idx;
2668     record_stream *s;
2669
2670     connection_assert_ref(c);
2671     pa_assert(t);
2672
2673     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2674         !pa_tagstruct_eof(t)) {
2675         protocol_error(c);
2676         return;
2677     }
2678
2679     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2680     s = pa_idxset_get_by_index(c->record_streams, idx);
2681     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2682
2683     pa_memblockq_flush(s->memblockq);
2684     pa_pstream_send_simple_ack(c->pstream, tag);
2685 }
2686
2687 static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2688     connection *c = CONNECTION(userdata);
2689     uint32_t idx;
2690     uint32_t maxlength, tlength, prebuf, minreq, fragsize;
2691     pa_tagstruct *reply;
2692
2693     connection_assert_ref(c);
2694     pa_assert(t);
2695
2696     if (pa_tagstruct_getu32(t, &idx) < 0) {
2697         protocol_error(c);
2698         return;
2699     }
2700
2701     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2702
2703     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR) {
2704         playback_stream *s;
2705
2706         s = pa_idxset_get_by_index(c->output_streams, idx);
2707         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2708         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2709
2710         if (pa_tagstruct_get(
2711                     t,
2712                     PA_TAG_U32, &maxlength,
2713                     PA_TAG_U32, &tlength,
2714                     PA_TAG_U32, &prebuf,
2715                     PA_TAG_U32, &minreq,
2716                     PA_TAG_INVALID) < 0 ||
2717             !pa_tagstruct_eof(t)) {
2718             protocol_error(c);
2719             return;
2720         }
2721
2722         CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
2723         CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
2724
2725         pa_memblockq_set_maxlength(s->memblockq, maxlength);
2726         pa_memblockq_set_tlength(s->memblockq, tlength);
2727         pa_memblockq_set_prebuf(s->memblockq, prebuf);
2728         pa_memblockq_set_minreq(s->memblockq, minreq);
2729
2730         reply = reply_new(tag);
2731         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_maxlength(s->memblockq));
2732         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_tlength(s->memblockq));
2733         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_prebuf(s->memblockq));
2734         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_minreq(s->memblockq));
2735
2736     } else {
2737         record_stream *s;
2738         size_t base;
2739         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR);
2740
2741         s = pa_idxset_get_by_index(c->record_streams, idx);
2742         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2743
2744         if (pa_tagstruct_get(
2745                     t,
2746                     PA_TAG_U32, &maxlength,
2747                     PA_TAG_U32, &fragsize,
2748                     PA_TAG_INVALID) < 0 ||
2749             !pa_tagstruct_eof(t)) {
2750             protocol_error(c);
2751             return;
2752         }
2753
2754         CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
2755         CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
2756
2757         pa_memblockq_set_maxlength(s->memblockq, maxlength);
2758
2759         base = pa_frame_size(&s->source_output->sample_spec);
2760         s->fragment_size = (fragsize/base)*base;
2761         if (s->fragment_size <= 0)
2762             s->fragment_size = base;
2763
2764         if (s->fragment_size > pa_memblockq_get_maxlength(s->memblockq))
2765             s->fragment_size = pa_memblockq_get_maxlength(s->memblockq);
2766
2767         reply = reply_new(tag);
2768         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_maxlength(s->memblockq));
2769         pa_tagstruct_putu32(reply, s->fragment_size);
2770     }
2771
2772     pa_pstream_send_tagstruct(c->pstream, reply);
2773 }
2774
2775 static void command_update_stream_sample_rate(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2776     connection *c = CONNECTION(userdata);
2777     uint32_t idx;
2778     uint32_t rate;
2779
2780     connection_assert_ref(c);
2781     pa_assert(t);
2782
2783     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2784         pa_tagstruct_getu32(t, &rate) < 0 ||
2785         !pa_tagstruct_eof(t)) {
2786         protocol_error(c);
2787         return;
2788     }
2789
2790     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2791     CHECK_VALIDITY(c->pstream, rate > 0 && rate <= PA_RATE_MAX, tag, PA_ERR_INVALID);
2792
2793     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE) {
2794         playback_stream *s;
2795
2796         s = pa_idxset_get_by_index(c->output_streams, idx);
2797         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2798         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2799
2800         pa_sink_input_set_rate(s->sink_input, rate);
2801
2802     } else {
2803         record_stream *s;
2804         pa_assert(command == PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE);
2805
2806         s = pa_idxset_get_by_index(c->record_streams, idx);
2807         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2808
2809         pa_source_output_set_rate(s->source_output, rate);
2810     }
2811
2812     pa_pstream_send_simple_ack(c->pstream, tag);
2813 }
2814
2815 static void command_update_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2816     connection *c = CONNECTION(userdata);
2817     uint32_t idx;
2818     uint32_t mode;
2819     pa_proplist *p;
2820
2821     connection_assert_ref(c);
2822     pa_assert(t);
2823
2824     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2825
2826     p = pa_proplist_new();
2827
2828     if (command == PA_COMMAND_UPDATE_CLIENT_PROPLIST) {
2829
2830         if (pa_tagstruct_getu32(t, &mode) < 0 ||
2831             pa_tagstruct_get_proplist(t, p) < 0 ||
2832             !pa_tagstruct_eof(t)) {
2833             protocol_error(c);
2834             pa_proplist_free(p);
2835             return;
2836         }
2837
2838     } else {
2839
2840         if (pa_tagstruct_getu32(t, &idx) < 0 ||
2841             pa_tagstruct_getu32(t, &mode) < 0 ||
2842             pa_tagstruct_get_proplist(t, p) < 0 ||
2843             !pa_tagstruct_eof(t)) {
2844             protocol_error(c);
2845             pa_proplist_free(p);
2846             return;
2847         }
2848     }
2849
2850     CHECK_VALIDITY(c->pstream, mode == PA_UPDATE_SET || mode == PA_UPDATE_MERGE || mode == PA_UPDATE_REPLACE, tag, PA_ERR_INVALID);
2851
2852     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST) {
2853         playback_stream *s;
2854
2855         s = pa_idxset_get_by_index(c->output_streams, idx);
2856         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2857         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2858
2859         pa_proplist_update(s->sink_input->proplist, mode, p);
2860         pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->sink_input->index);
2861
2862     } else if (command == PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST) {
2863         record_stream *s;
2864
2865         s = pa_idxset_get_by_index(c->record_streams, idx);
2866         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2867
2868         pa_proplist_update(s->source_output->proplist, mode, p);
2869         pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->source_output->index);
2870     } else {
2871         pa_assert(command == PA_COMMAND_UPDATE_CLIENT_PROPLIST);
2872
2873         pa_proplist_update(c->client->proplist, mode, p);
2874         pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_CLIENT|PA_SUBSCRIPTION_EVENT_CHANGE, c->client->index);
2875     }
2876
2877     pa_pstream_send_simple_ack(c->pstream, tag);
2878 }
2879
2880 static void command_remove_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2881     connection *c = CONNECTION(userdata);
2882     uint32_t idx;
2883     unsigned changed = 0;
2884     pa_proplist *p;
2885     pa_strlist *l = NULL;
2886
2887     connection_assert_ref(c);
2888     pa_assert(t);
2889
2890     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2891
2892     if (command != PA_COMMAND_REMOVE_CLIENT_PROPLIST) {
2893
2894         if (pa_tagstruct_getu32(t, &idx) < 0) {
2895             protocol_error(c);
2896             return;
2897         }
2898     }
2899
2900     if (command == PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST) {
2901         playback_stream *s;
2902
2903         s = pa_idxset_get_by_index(c->output_streams, idx);
2904         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2905         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2906
2907         p = s->sink_input->proplist;
2908
2909     } else if (command == PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST) {
2910         record_stream *s;
2911
2912         s = pa_idxset_get_by_index(c->record_streams, idx);
2913         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2914
2915         p = s->source_output->proplist;
2916     } else {
2917         pa_assert(command == PA_COMMAND_REMOVE_CLIENT_PROPLIST);
2918
2919         p = c->client->proplist;
2920     }
2921
2922     for (;;) {
2923         const char *k;
2924
2925         if (pa_tagstruct_gets(t, &k) < 0) {
2926             protocol_error(c);
2927             pa_strlist_free(l);
2928             return;
2929         }
2930
2931         if (!k)
2932             break;
2933
2934         l = pa_strlist_prepend(l, k);
2935     }
2936
2937     if (!pa_tagstruct_eof(t)) {
2938         protocol_error(c);
2939         pa_strlist_free(l);
2940         return;
2941     }
2942
2943     for (;;) {
2944         char *z;
2945
2946         l = pa_strlist_pop(l, &z);
2947
2948         if (!z)
2949             break;
2950
2951         changed += pa_proplist_unset(p, z) >= 0;
2952         pa_xfree(z);
2953     }
2954
2955     pa_pstream_send_simple_ack(c->pstream, tag);
2956
2957     if (changed) {
2958         if (command == PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST) {
2959             playback_stream *s;
2960
2961             s = pa_idxset_get_by_index(c->output_streams, idx);
2962             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->sink_input->index);
2963
2964         } else if (command == PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST) {
2965             record_stream *s;
2966
2967             s = pa_idxset_get_by_index(c->record_streams, idx);
2968             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->source_output->index);
2969
2970         } else {
2971             pa_assert(command == PA_COMMAND_REMOVE_CLIENT_PROPLIST);
2972             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_CLIENT|PA_SUBSCRIPTION_EVENT_CHANGE, c->client->index);
2973         }
2974     }
2975 }
2976
2977 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) {
2978     connection *c = CONNECTION(userdata);
2979     const char *s;
2980
2981     connection_assert_ref(c);
2982     pa_assert(t);
2983
2984     if (pa_tagstruct_gets(t, &s) < 0 ||
2985         !pa_tagstruct_eof(t)) {
2986         protocol_error(c);
2987         return;
2988     }
2989
2990     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2991     CHECK_VALIDITY(c->pstream, !s || (*s && pa_utf8_valid(s)), tag, PA_ERR_INVALID);
2992
2993     pa_namereg_set_default(c->protocol->core, s, command == PA_COMMAND_SET_DEFAULT_SOURCE ? PA_NAMEREG_SOURCE : PA_NAMEREG_SINK);
2994     pa_pstream_send_simple_ack(c->pstream, tag);
2995 }
2996
2997 static void command_set_stream_name(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2998     connection *c = CONNECTION(userdata);
2999     uint32_t idx;
3000     const char *name;
3001
3002     connection_assert_ref(c);
3003     pa_assert(t);
3004
3005     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3006         pa_tagstruct_gets(t, &name) < 0 ||
3007         !pa_tagstruct_eof(t)) {
3008         protocol_error(c);
3009         return;
3010     }
3011
3012     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3013     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
3014
3015     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
3016         playback_stream *s;
3017
3018         s = pa_idxset_get_by_index(c->output_streams, idx);
3019         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3020         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3021
3022         pa_sink_input_set_name(s->sink_input, name);
3023
3024     } else {
3025         record_stream *s;
3026         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_NAME);
3027
3028         s = pa_idxset_get_by_index(c->record_streams, idx);
3029         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3030
3031         pa_source_output_set_name(s->source_output, name);
3032     }
3033
3034     pa_pstream_send_simple_ack(c->pstream, tag);
3035 }
3036
3037 static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3038     connection *c = CONNECTION(userdata);
3039     uint32_t idx;
3040
3041     connection_assert_ref(c);
3042     pa_assert(t);
3043
3044     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3045         !pa_tagstruct_eof(t)) {
3046         protocol_error(c);
3047         return;
3048     }
3049
3050     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3051
3052     if (command == PA_COMMAND_KILL_CLIENT) {
3053         pa_client *client;
3054
3055         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
3056         CHECK_VALIDITY(c->pstream, client, tag, PA_ERR_NOENTITY);
3057
3058         connection_ref(c);
3059         pa_client_kill(client);
3060
3061     } else if (command == PA_COMMAND_KILL_SINK_INPUT) {
3062         pa_sink_input *s;
3063
3064         s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3065         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3066
3067         connection_ref(c);
3068         pa_sink_input_kill(s);
3069     } else {
3070         pa_source_output *s;
3071
3072         pa_assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
3073
3074         s = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
3075         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3076
3077         connection_ref(c);
3078         pa_source_output_kill(s);
3079     }
3080
3081     pa_pstream_send_simple_ack(c->pstream, tag);
3082     connection_unref(c);
3083 }
3084
3085 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) {
3086     connection *c = CONNECTION(userdata);
3087     pa_module *m;
3088     const char *name, *argument;
3089     pa_tagstruct *reply;
3090
3091     connection_assert_ref(c);
3092     pa_assert(t);
3093
3094     if (pa_tagstruct_gets(t, &name) < 0 ||
3095         pa_tagstruct_gets(t, &argument) < 0 ||
3096         !pa_tagstruct_eof(t)) {
3097         protocol_error(c);
3098         return;
3099     }
3100
3101     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3102     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name) && !strchr(name, '/'), tag, PA_ERR_INVALID);
3103     CHECK_VALIDITY(c->pstream, !argument || pa_utf8_valid(argument), tag, PA_ERR_INVALID);
3104
3105     if (!(m = pa_module_load(c->protocol->core, name, argument))) {
3106         pa_pstream_send_error(c->pstream, tag, PA_ERR_MODINITFAILED);
3107         return;
3108     }
3109
3110     reply = reply_new(tag);
3111     pa_tagstruct_putu32(reply, m->index);
3112     pa_pstream_send_tagstruct(c->pstream, reply);
3113 }
3114
3115 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) {
3116     connection *c = CONNECTION(userdata);
3117     uint32_t idx;
3118     pa_module *m;
3119
3120     connection_assert_ref(c);
3121     pa_assert(t);
3122
3123     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3124         !pa_tagstruct_eof(t)) {
3125         protocol_error(c);
3126         return;
3127     }
3128
3129     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3130     m = pa_idxset_get_by_index(c->protocol->core->modules, idx);
3131     CHECK_VALIDITY(c->pstream, m, tag, PA_ERR_NOENTITY);
3132
3133     pa_module_unload_request(m);
3134     pa_pstream_send_simple_ack(c->pstream, tag);
3135 }
3136
3137 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) {
3138     connection *c = CONNECTION(userdata);
3139     const char *name, *module, *argument;
3140     uint32_t type;
3141     uint32_t idx;
3142     pa_tagstruct *reply;
3143
3144     connection_assert_ref(c);
3145     pa_assert(t);
3146
3147     if (pa_tagstruct_gets(t, &name) < 0 ||
3148         pa_tagstruct_getu32(t, &type) < 0 ||
3149         pa_tagstruct_gets(t, &module) < 0 ||
3150         pa_tagstruct_gets(t, &argument) < 0 ||
3151         !pa_tagstruct_eof(t)) {
3152         protocol_error(c);
3153         return;
3154     }
3155
3156     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3157     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
3158     CHECK_VALIDITY(c->pstream, type == 0 || type == 1, tag, PA_ERR_INVALID);
3159     CHECK_VALIDITY(c->pstream, module && *module && pa_utf8_valid(module), tag, PA_ERR_INVALID);
3160     CHECK_VALIDITY(c->pstream, !argument || pa_utf8_valid(argument), tag, PA_ERR_INVALID);
3161
3162     if (pa_autoload_add(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE, module, argument, &idx) < 0) {
3163         pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
3164         return;
3165     }
3166
3167     reply = reply_new(tag);
3168     pa_tagstruct_putu32(reply, idx);
3169     pa_pstream_send_tagstruct(c->pstream, reply);
3170 }
3171
3172 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) {
3173     connection *c = CONNECTION(userdata);
3174     const char *name = NULL;
3175     uint32_t type, idx = PA_IDXSET_INVALID;
3176     int r;
3177
3178     connection_assert_ref(c);
3179     pa_assert(t);
3180
3181     if ((pa_tagstruct_getu32(t, &idx) < 0 &&
3182         (pa_tagstruct_gets(t, &name) < 0 ||
3183          pa_tagstruct_getu32(t, &type) < 0)) ||
3184         !pa_tagstruct_eof(t)) {
3185         protocol_error(c);
3186         return;
3187     }
3188
3189     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3190     CHECK_VALIDITY(c->pstream, name || idx != PA_IDXSET_INVALID, tag, PA_ERR_INVALID);
3191     CHECK_VALIDITY(c->pstream, !name || (*name && pa_utf8_valid(name) && (type == 0 || type == 1)), tag, PA_ERR_INVALID);
3192
3193     if (name)
3194         r = pa_autoload_remove_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
3195     else
3196         r = pa_autoload_remove_by_index(c->protocol->core, idx);
3197
3198     CHECK_VALIDITY(c->pstream, r >= 0, tag, PA_ERR_NOENTITY);
3199
3200     pa_pstream_send_simple_ack(c->pstream, tag);
3201 }
3202
3203 static void autoload_fill_tagstruct(pa_tagstruct *t, const pa_autoload_entry *e) {
3204     pa_assert(t && e);
3205
3206     pa_tagstruct_putu32(t, e->index);
3207     pa_tagstruct_puts(t, e->name);
3208     pa_tagstruct_putu32(t, e->type == PA_NAMEREG_SINK ? 0 : 1);
3209     pa_tagstruct_puts(t, e->module);
3210     pa_tagstruct_puts(t, e->argument);
3211 }
3212
3213 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) {
3214     connection *c = CONNECTION(userdata);
3215     const pa_autoload_entry *a = NULL;
3216     uint32_t type, idx;
3217     const char *name;
3218     pa_tagstruct *reply;
3219
3220     connection_assert_ref(c);
3221     pa_assert(t);
3222
3223     if ((pa_tagstruct_getu32(t, &idx) < 0 &&
3224         (pa_tagstruct_gets(t, &name) < 0 ||
3225          pa_tagstruct_getu32(t, &type) < 0)) ||
3226         !pa_tagstruct_eof(t)) {
3227         protocol_error(c);
3228         return;
3229     }
3230
3231     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3232     CHECK_VALIDITY(c->pstream, name || idx != PA_IDXSET_INVALID, tag, PA_ERR_INVALID);
3233     CHECK_VALIDITY(c->pstream, !name || (*name && (type == 0 || type == 1) && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
3234
3235     if (name)
3236         a = pa_autoload_get_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
3237     else
3238         a = pa_autoload_get_by_index(c->protocol->core, idx);
3239
3240     CHECK_VALIDITY(c->pstream, a, tag, PA_ERR_NOENTITY);
3241
3242     reply = reply_new(tag);
3243     autoload_fill_tagstruct(reply, a);
3244     pa_pstream_send_tagstruct(c->pstream, reply);
3245 }
3246
3247 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) {
3248     connection *c = CONNECTION(userdata);
3249     pa_tagstruct *reply;
3250
3251     connection_assert_ref(c);
3252     pa_assert(t);
3253
3254     if (!pa_tagstruct_eof(t)) {
3255         protocol_error(c);
3256         return;
3257     }
3258
3259     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3260
3261     reply = reply_new(tag);
3262
3263     if (c->protocol->core->autoload_hashmap) {
3264         pa_autoload_entry *a;
3265         void *state = NULL;
3266
3267         while ((a = pa_hashmap_iterate(c->protocol->core->autoload_hashmap, &state, NULL)))
3268             autoload_fill_tagstruct(reply, a);
3269     }
3270
3271     pa_pstream_send_tagstruct(c->pstream, reply);
3272 }
3273
3274 static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3275     connection *c = CONNECTION(userdata);
3276     uint32_t idx = PA_INVALID_INDEX, idx_device = PA_INVALID_INDEX;
3277     const char *name = NULL;
3278
3279     connection_assert_ref(c);
3280     pa_assert(t);
3281
3282     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3283         pa_tagstruct_getu32(t, &idx_device) < 0 ||
3284         pa_tagstruct_gets(t, &name) < 0 ||
3285         !pa_tagstruct_eof(t)) {
3286         protocol_error(c);
3287         return;
3288     }
3289
3290     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3291     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3292     CHECK_VALIDITY(c->pstream, idx_device != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
3293
3294     if (command == PA_COMMAND_MOVE_SINK_INPUT) {
3295         pa_sink_input *si = NULL;
3296         pa_sink *sink = NULL;
3297
3298         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3299
3300         if (idx_device != PA_INVALID_INDEX)
3301             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx_device);
3302         else
3303             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
3304
3305         CHECK_VALIDITY(c->pstream, si && sink, tag, PA_ERR_NOENTITY);
3306
3307         if (pa_sink_input_move_to(si, sink, 0) < 0) {
3308             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3309             return;
3310         }
3311     } else {
3312         pa_source_output *so = NULL;
3313         pa_source *source;
3314
3315         pa_assert(command == PA_COMMAND_MOVE_SOURCE_OUTPUT);
3316
3317         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
3318
3319         if (idx_device != PA_INVALID_INDEX)
3320             source = pa_idxset_get_by_index(c->protocol->core->sources, idx_device);
3321         else
3322             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
3323
3324         CHECK_VALIDITY(c->pstream, so && source, tag, PA_ERR_NOENTITY);
3325
3326         if (pa_source_output_move_to(so, source) < 0) {
3327             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3328             return;
3329         }
3330     }
3331
3332     pa_pstream_send_simple_ack(c->pstream, tag);
3333 }
3334
3335 static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3336     connection *c = CONNECTION(userdata);
3337     uint32_t idx = PA_INVALID_INDEX;
3338     const char *name = NULL;
3339     int b;
3340
3341     connection_assert_ref(c);
3342     pa_assert(t);
3343
3344     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3345         pa_tagstruct_gets(t, &name) < 0 ||
3346         pa_tagstruct_get_boolean(t, &b) < 0 ||
3347         !pa_tagstruct_eof(t)) {
3348         protocol_error(c);
3349         return;
3350     }
3351
3352     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3353     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || !*name || pa_utf8_valid(name), tag, PA_ERR_INVALID);
3354
3355     if (command == PA_COMMAND_SUSPEND_SINK) {
3356
3357         if (idx == PA_INVALID_INDEX && name && !*name) {
3358
3359             if (pa_sink_suspend_all(c->protocol->core, b) < 0) {
3360                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3361                 return;
3362             }
3363         } else {
3364             pa_sink *sink = NULL;
3365
3366             if (idx != PA_INVALID_INDEX)
3367                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3368             else
3369                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
3370
3371             CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
3372
3373             if (pa_sink_suspend(sink, b) < 0) {
3374                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3375                 return;
3376             }
3377         }
3378     } else {
3379
3380         pa_assert(command == PA_COMMAND_SUSPEND_SOURCE);
3381
3382         if (idx == PA_INVALID_INDEX && name && !*name) {
3383
3384             if (pa_source_suspend_all(c->protocol->core, b) < 0) {
3385                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3386                 return;
3387             }
3388
3389         } else {
3390             pa_source *source;
3391
3392             if (idx != PA_INVALID_INDEX)
3393                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3394             else
3395                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
3396
3397             CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
3398
3399             if (pa_source_suspend(source, b) < 0) {
3400                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3401                 return;
3402             }
3403         }
3404     }
3405
3406     pa_pstream_send_simple_ack(c->pstream, tag);
3407 }
3408
3409 /*** pstream callbacks ***/
3410
3411 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
3412     connection *c = CONNECTION(userdata);
3413
3414     pa_assert(p);
3415     pa_assert(packet);
3416     connection_assert_ref(c);
3417
3418     if (pa_pdispatch_run(c->pdispatch, packet, creds, c) < 0) {
3419         pa_log("invalid packet.");
3420         connection_unlink(c);
3421     }
3422 }
3423
3424 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) {
3425     connection *c = CONNECTION(userdata);
3426     output_stream *stream;
3427
3428     pa_assert(p);
3429     pa_assert(chunk);
3430     connection_assert_ref(c);
3431
3432     if (!(stream = OUTPUT_STREAM(pa_idxset_get_by_index(c->output_streams, channel)))) {
3433         pa_log("client sent block for invalid stream.");
3434         /* Ignoring */
3435         return;
3436     }
3437
3438     if (playback_stream_isinstance(stream)) {
3439         playback_stream *ps = PLAYBACK_STREAM(stream);
3440
3441         if (seek != PA_SEEK_RELATIVE || offset != 0)
3442             pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_SEEK, PA_UINT_TO_PTR(seek), offset, NULL, NULL);
3443
3444         pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
3445
3446     } else {
3447         upload_stream *u = UPLOAD_STREAM(stream);
3448         size_t l;
3449
3450         if (!u->memchunk.memblock) {
3451             if (u->length == chunk->length) {
3452                 u->memchunk = *chunk;
3453                 pa_memblock_ref(u->memchunk.memblock);
3454                 u->length = 0;
3455             } else {
3456                 u->memchunk.memblock = pa_memblock_new(c->protocol->core->mempool, u->length);
3457                 u->memchunk.index = u->memchunk.length = 0;
3458             }
3459         }
3460
3461         pa_assert(u->memchunk.memblock);
3462
3463         l = u->length;
3464         if (l > chunk->length)
3465             l = chunk->length;
3466
3467
3468         if (l > 0) {
3469             void *src, *dst;
3470             dst = pa_memblock_acquire(u->memchunk.memblock);
3471             src = pa_memblock_acquire(chunk->memblock);
3472
3473             memcpy((uint8_t*) dst + u->memchunk.index + u->memchunk.length,
3474                    (uint8_t*) src+chunk->index, l);
3475
3476             pa_memblock_release(u->memchunk.memblock);
3477             pa_memblock_release(chunk->memblock);
3478
3479             u->memchunk.length += l;
3480             u->length -= l;
3481         }
3482     }
3483 }
3484
3485 static void pstream_die_callback(pa_pstream *p, void *userdata) {
3486     connection *c = CONNECTION(userdata);
3487
3488     pa_assert(p);
3489     connection_assert_ref(c);
3490
3491     connection_unlink(c);
3492     pa_log_info("connection died.");
3493 }
3494
3495 static void pstream_drain_callback(pa_pstream *p, void *userdata) {
3496     connection *c = CONNECTION(userdata);
3497
3498     pa_assert(p);
3499     connection_assert_ref(c);
3500
3501     send_memblock(c);
3502 }
3503
3504 static void pstream_revoke_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
3505     pa_thread_mq *q;
3506
3507     if (!(q = pa_thread_mq_get()))
3508         pa_pstream_send_revoke(p, block_id);
3509     else
3510         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_REVOKE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
3511 }
3512
3513 static void pstream_release_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
3514     pa_thread_mq *q;
3515
3516     if (!(q = pa_thread_mq_get()))
3517         pa_pstream_send_release(p, block_id);
3518     else
3519         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_RELEASE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
3520 }
3521
3522 /*** client callbacks ***/
3523
3524 static void client_kill_cb(pa_client *c) {
3525     pa_assert(c);
3526
3527     connection_unlink(CONNECTION(c->userdata));
3528 }
3529
3530 /*** socket server callbacks ***/
3531
3532 static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *tv, void *userdata) {
3533     connection *c = CONNECTION(userdata);
3534
3535     pa_assert(m);
3536     pa_assert(tv);
3537     connection_assert_ref(c);
3538     pa_assert(c->auth_timeout_event == e);
3539
3540     if (!c->authorized)
3541         connection_unlink(c);
3542 }
3543
3544 static void on_connection(PA_GCC_UNUSED pa_socket_server*s, pa_iochannel *io, void *userdata) {
3545     pa_protocol_native *p = userdata;
3546     connection *c;
3547     char cname[256], pname[128];
3548
3549     pa_assert(s);
3550     pa_assert(io);
3551     pa_assert(p);
3552
3553     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
3554         pa_log_warn("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
3555         pa_iochannel_free(io);
3556         return;
3557     }
3558
3559     c = pa_msgobject_new(connection);
3560     c->parent.parent.free = connection_free;
3561     c->parent.process_msg = connection_process_msg;
3562
3563     c->authorized = p->public;
3564
3565     if (!c->authorized && p->auth_ip_acl && pa_ip_acl_check(p->auth_ip_acl, pa_iochannel_get_recv_fd(io)) > 0) {
3566         pa_log_info("Client authenticated by IP ACL.");
3567         c->authorized = TRUE;
3568     }
3569
3570     if (!c->authorized) {
3571         struct timeval tv;
3572         pa_gettimeofday(&tv);
3573         tv.tv_sec += AUTH_TIMEOUT;
3574         c->auth_timeout_event = p->core->mainloop->time_new(p->core->mainloop, &tv, auth_timeout, c);
3575     } else
3576         c->auth_timeout_event = NULL;
3577
3578     c->version = 8;
3579     c->protocol = p;
3580     pa_iochannel_socket_peer_to_string(io, pname, sizeof(pname));
3581     pa_snprintf(cname, sizeof(cname), "Native client (%s)", pname);
3582     c->client = pa_client_new(p->core, __FILE__, cname);
3583     c->client->kill = client_kill_cb;
3584     c->client->userdata = c;
3585     c->client->module = p->module;
3586
3587     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->mempool);
3588
3589     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
3590     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
3591     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
3592     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
3593     pa_pstream_set_revoke_callback(c->pstream, pstream_revoke_callback, c);
3594     pa_pstream_set_release_callback(c->pstream, pstream_release_callback, c);
3595
3596     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
3597
3598     c->record_streams = pa_idxset_new(NULL, NULL);
3599     c->output_streams = pa_idxset_new(NULL, NULL);
3600
3601     c->rrobin_index = PA_IDXSET_INVALID;
3602     c->subscription = NULL;
3603
3604     pa_idxset_put(p->connections, c, NULL);
3605
3606 #ifdef HAVE_CREDS
3607     if (pa_iochannel_creds_supported(io))
3608         pa_iochannel_creds_enable(io);
3609
3610 #endif
3611 }
3612
3613 /*** module entry points ***/
3614
3615 static int load_key(pa_protocol_native*p, const char*fn) {
3616     pa_assert(p);
3617
3618     p->auth_cookie_in_property = FALSE;
3619
3620     if (!fn && pa_authkey_prop_get(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0) {
3621         pa_log_info("using already loaded auth cookie.");
3622         pa_authkey_prop_ref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
3623         p->auth_cookie_in_property = TRUE;
3624         return 0;
3625     }
3626
3627     if (!fn)
3628         fn = PA_NATIVE_COOKIE_FILE;
3629
3630     if (pa_authkey_load_auto(fn, p->auth_cookie, sizeof(p->auth_cookie)) < 0)
3631         return -1;
3632
3633     pa_log_info("loading cookie from disk.");
3634
3635     if (pa_authkey_prop_put(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0)
3636         p->auth_cookie_in_property = TRUE;
3637
3638     return 0;
3639 }
3640
3641 static pa_protocol_native* protocol_new_internal(pa_core *c, pa_module *m, pa_modargs *ma) {
3642     pa_protocol_native *p;
3643     pa_bool_t public = FALSE;
3644     const char *acl;
3645
3646     pa_assert(c);
3647     pa_assert(ma);
3648
3649     if (pa_modargs_get_value_boolean(ma, "auth-anonymous", &public) < 0) {
3650         pa_log("auth-anonymous= expects a boolean argument.");
3651         return NULL;
3652     }
3653
3654     p = pa_xnew(pa_protocol_native, 1);
3655     p->core = c;
3656     p->module = m;
3657     p->public = public;
3658     p->server = NULL;
3659     p->auth_ip_acl = NULL;
3660
3661 #ifdef HAVE_CREDS
3662     {
3663         pa_bool_t a = 1;
3664         if (pa_modargs_get_value_boolean(ma, "auth-group-enabled", &a) < 0) {
3665             pa_log("auth-group-enabled= expects a boolean argument.");
3666             return NULL;
3667         }
3668         p->auth_group = a ? pa_xstrdup(pa_modargs_get_value(ma, "auth-group", c->is_system_instance ? PA_ACCESS_GROUP : NULL)) : NULL;
3669
3670         if (p->auth_group)
3671             pa_log_info("Allowing access to group '%s'.", p->auth_group);
3672     }
3673 #endif
3674
3675
3676     if ((acl = pa_modargs_get_value(ma, "auth-ip-acl", NULL))) {
3677
3678         if (!(p->auth_ip_acl = pa_ip_acl_new(acl))) {
3679             pa_log("Failed to parse IP ACL '%s'", acl);
3680             goto fail;
3681         }
3682     }
3683
3684     if (load_key(p, pa_modargs_get_value(ma, "cookie", NULL)) < 0)
3685         goto fail;
3686
3687     p->connections = pa_idxset_new(NULL, NULL);
3688
3689     return p;
3690
3691 fail:
3692 #ifdef HAVE_CREDS
3693     pa_xfree(p->auth_group);
3694 #endif
3695     if (p->auth_ip_acl)
3696         pa_ip_acl_free(p->auth_ip_acl);
3697     pa_xfree(p);
3698     return NULL;
3699 }
3700
3701 pa_protocol_native* pa_protocol_native_new(pa_core *core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
3702     char t[256];
3703     pa_protocol_native *p;
3704
3705     if (!(p = protocol_new_internal(core, m, ma)))
3706         return NULL;
3707
3708     p->server = server;
3709     pa_socket_server_set_callback(p->server, on_connection, p);
3710
3711     if (pa_socket_server_get_address(p->server, t, sizeof(t))) {
3712         pa_strlist *l;
3713         l = pa_property_get(core, PA_NATIVE_SERVER_PROPERTY_NAME);
3714         l = pa_strlist_prepend(l, t);
3715         pa_property_replace(core, PA_NATIVE_SERVER_PROPERTY_NAME, l);
3716     }
3717
3718     return p;
3719 }
3720
3721 void pa_protocol_native_free(pa_protocol_native *p) {
3722     connection *c;
3723     pa_assert(p);
3724
3725     while ((c = pa_idxset_first(p->connections, NULL)))
3726         connection_unlink(c);
3727     pa_idxset_free(p->connections, NULL, NULL);
3728
3729     if (p->server) {
3730         char t[256];
3731
3732         if (pa_socket_server_get_address(p->server, t, sizeof(t))) {
3733             pa_strlist *l;
3734             l = pa_property_get(p->core, PA_NATIVE_SERVER_PROPERTY_NAME);
3735             l = pa_strlist_remove(l, t);
3736
3737             if (l)
3738                 pa_property_replace(p->core, PA_NATIVE_SERVER_PROPERTY_NAME, l);
3739             else
3740                 pa_property_remove(p->core, PA_NATIVE_SERVER_PROPERTY_NAME);
3741         }
3742
3743         pa_socket_server_unref(p->server);
3744     }
3745
3746     if (p->auth_cookie_in_property)
3747         pa_authkey_prop_unref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
3748
3749     if (p->auth_ip_acl)
3750         pa_ip_acl_free(p->auth_ip_acl);
3751
3752 #ifdef HAVE_CREDS
3753     pa_xfree(p->auth_group);
3754 #endif
3755     pa_xfree(p);
3756 }
3757
3758 pa_protocol_native* pa_protocol_native_new_iochannel(
3759         pa_core*core,
3760         pa_iochannel *io,
3761         pa_module *m,
3762         pa_modargs *ma) {
3763
3764     pa_protocol_native *p;
3765
3766     if (!(p = protocol_new_internal(core, m, ma)))
3767         return NULL;
3768
3769     on_connection(NULL, io, p);
3770
3771     return p;
3772 }