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