src/pulsecore/protocol-native.c

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