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