code.delx.au - pulseaudio/blob - src/pulsecore/protocol-native.c

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