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

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