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

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