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