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