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
1356     pa_sink_input_assert_ref(i);
1357     s = PLAYBACK_STREAM(i->userdata);
1358     playback_stream_assert_ref(s);
1359
1360     switch (code) {
1361
1362         case SINK_INPUT_MESSAGE_SEEK: {
1363             int64_t windex;
1364
1365             windex = 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
1373             handle_seek(s, windex);
1374             return 0;
1375         }
1376
1377         case SINK_INPUT_MESSAGE_POST_DATA: {
1378             int64_t windex;
1379
1380             pa_assert(chunk);
1381
1382             windex = pa_memblockq_get_write_index(s->memblockq);
1383
1384 /*             pa_log("sink input post: %lu %lli", (unsigned long) chunk->length, (long long) windex); */
1385
1386             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
1387
1388                 if (pa_log_ratelimit(PA_LOG_WARN))
1389                     pa_log_warn("Failed to push data into queue");
1390                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_OVERFLOW, NULL, 0, NULL, NULL);
1391                 pa_memblockq_seek(s->memblockq, (int64_t) chunk->length, PA_SEEK_RELATIVE, TRUE);
1392             }
1393
1394             handle_seek(s, windex);
1395
1396 /*             pa_log("sink input post2: %lu", (unsigned long) pa_memblockq_get_length(s->memblockq)); */
1397
1398             return 0;
1399         }
1400
1401         case SINK_INPUT_MESSAGE_DRAIN:
1402         case SINK_INPUT_MESSAGE_FLUSH:
1403         case SINK_INPUT_MESSAGE_PREBUF_FORCE:
1404         case SINK_INPUT_MESSAGE_TRIGGER: {
1405
1406             int64_t windex;
1407             pa_sink_input *isync;
1408             void (*func)(pa_memblockq *bq);
1409
1410             switch  (code) {
1411                 case SINK_INPUT_MESSAGE_FLUSH:
1412                     func = flush_write_no_account;
1413                     break;
1414
1415                 case SINK_INPUT_MESSAGE_PREBUF_FORCE:
1416                     func = pa_memblockq_prebuf_force;
1417                     break;
1418
1419                 case SINK_INPUT_MESSAGE_DRAIN:
1420                 case SINK_INPUT_MESSAGE_TRIGGER:
1421                     func = pa_memblockq_prebuf_disable;
1422                     break;
1423
1424                 default:
1425                     pa_assert_not_reached();
1426             }
1427
1428             windex = pa_memblockq_get_write_index(s->memblockq);
1429             func(s->memblockq);
1430             handle_seek(s, windex);
1431
1432             /* Do the same for all other members in the sync group */
1433             for (isync = i->sync_prev; isync; isync = isync->sync_prev) {
1434                 playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
1435                 windex = pa_memblockq_get_write_index(ssync->memblockq);
1436                 func(ssync->memblockq);
1437                 handle_seek(ssync, windex);
1438             }
1439
1440             for (isync = i->sync_next; isync; isync = isync->sync_next) {
1441                 playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
1442                 windex = pa_memblockq_get_write_index(ssync->memblockq);
1443                 func(ssync->memblockq);
1444                 handle_seek(ssync, windex);
1445             }
1446
1447             if (code == SINK_INPUT_MESSAGE_DRAIN) {
1448                 if (!pa_memblockq_is_readable(s->memblockq))
1449                     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_DRAIN_ACK, userdata, 0, NULL, NULL);
1450                 else {
1451                     s->drain_tag = PA_PTR_TO_UINT(userdata);
1452                     s->drain_request = TRUE;
1453                 }
1454             }
1455
1456             return 0;
1457         }
1458
1459         case SINK_INPUT_MESSAGE_UPDATE_LATENCY:
1460             /* Atomically get a snapshot of all timing parameters... */
1461             s->read_index = pa_memblockq_get_read_index(s->memblockq);
1462             s->write_index = pa_memblockq_get_write_index(s->memblockq);
1463             s->render_memblockq_length = pa_memblockq_get_length(s->sink_input->thread_info.render_memblockq);
1464             s->current_sink_latency = pa_sink_get_latency_within_thread(s->sink_input->sink);
1465             s->underrun_for = s->sink_input->thread_info.underrun_for;
1466             s->playing_for = s->sink_input->thread_info.playing_for;
1467
1468             return 0;
1469
1470         case PA_SINK_INPUT_MESSAGE_SET_STATE: {
1471             int64_t windex;
1472
1473             windex = pa_memblockq_get_write_index(s->memblockq);
1474
1475             pa_memblockq_prebuf_force(s->memblockq);
1476
1477             handle_seek(s, windex);
1478
1479             /* Fall through to the default handler */
1480             break;
1481         }
1482
1483         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
1484             pa_usec_t *r = userdata;
1485
1486             *r = pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
1487
1488             /* Fall through, the default handler will add in the extra
1489              * latency added by the resampler */
1490             break;
1491         }
1492
1493         case SINK_INPUT_MESSAGE_UPDATE_BUFFER_ATTR: {
1494             pa_memblockq_apply_attr(s->memblockq, &s->buffer_attr);
1495             pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
1496             return 0;
1497         }
1498     }
1499
1500     return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
1501 }
1502
1503 /* Called from thread context */
1504 static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk) {
1505     playback_stream *s;
1506
1507     pa_sink_input_assert_ref(i);
1508     s = PLAYBACK_STREAM(i->userdata);
1509     playback_stream_assert_ref(s);
1510     pa_assert(chunk);
1511
1512 /*     pa_log("%s, pop(): %lu", pa_proplist_gets(i->proplist, PA_PROP_MEDIA_NAME), (unsigned long) pa_memblockq_get_length(s->memblockq)); */
1513
1514     if (pa_memblockq_is_readable(s->memblockq))
1515         s->is_underrun = FALSE;
1516     else {
1517         if (!s->is_underrun)
1518             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));
1519
1520         if (s->drain_request && pa_sink_input_safe_to_remove(i)) {
1521             s->drain_request = FALSE;
1522             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);
1523         } else if (!s->is_underrun)
1524             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_UNDERFLOW, NULL, 0, NULL, NULL);
1525
1526         s->is_underrun = TRUE;
1527
1528         playback_stream_request_bytes(s);
1529     }
1530
1531     /* This call will not fail with prebuf=0, hence we check for
1532        underrun explicitly above */
1533     if (pa_memblockq_peek(s->memblockq, chunk) < 0)
1534         return -1;
1535
1536     chunk->length = PA_MIN(nbytes, chunk->length);
1537
1538     if (i->thread_info.underrun_for > 0)
1539         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_STARTED, NULL, 0, NULL, NULL);
1540
1541     pa_memblockq_drop(s->memblockq, chunk->length);
1542     playback_stream_request_bytes(s);
1543
1544     return 0;
1545 }
1546
1547 /* Called from thread context */
1548 static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
1549     playback_stream *s;
1550
1551     pa_sink_input_assert_ref(i);
1552     s = PLAYBACK_STREAM(i->userdata);
1553     playback_stream_assert_ref(s);
1554
1555     /* If we are in an underrun, then we don't rewind */
1556     if (i->thread_info.underrun_for > 0)
1557         return;
1558
1559     pa_memblockq_rewind(s->memblockq, nbytes);
1560 }
1561
1562 /* Called from thread context */
1563 static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
1564     playback_stream *s;
1565
1566     pa_sink_input_assert_ref(i);
1567     s = PLAYBACK_STREAM(i->userdata);
1568     playback_stream_assert_ref(s);
1569
1570     pa_memblockq_set_maxrewind(s->memblockq, nbytes);
1571 }
1572
1573 /* Called from thread context */
1574 static void sink_input_update_max_request_cb(pa_sink_input *i, size_t nbytes) {
1575     playback_stream *s;
1576     size_t new_tlength, old_tlength;
1577
1578     pa_sink_input_assert_ref(i);
1579     s = PLAYBACK_STREAM(i->userdata);
1580     playback_stream_assert_ref(s);
1581
1582     old_tlength = pa_memblockq_get_tlength(s->memblockq);
1583     new_tlength = nbytes+2*pa_memblockq_get_minreq(s->memblockq);
1584
1585     if (old_tlength < new_tlength) {
1586         pa_log_debug("max_request changed, trying to update from %zu to %zu.", old_tlength, new_tlength);
1587         pa_memblockq_set_tlength(s->memblockq, new_tlength);
1588         new_tlength = pa_memblockq_get_tlength(s->memblockq);
1589
1590         if (new_tlength == old_tlength)
1591             pa_log_debug("Failed to increase tlength");
1592         else {
1593             pa_log_debug("Notifying client about increased tlength");
1594             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);
1595         }
1596     }
1597 }
1598
1599 /* Called from main context */
1600 static void sink_input_kill_cb(pa_sink_input *i) {
1601     playback_stream *s;
1602
1603     pa_sink_input_assert_ref(i);
1604     s = PLAYBACK_STREAM(i->userdata);
1605     playback_stream_assert_ref(s);
1606
1607     playback_stream_send_killed(s);
1608     playback_stream_unlink(s);
1609 }
1610
1611 /* Called from main context */
1612 static void sink_input_send_event_cb(pa_sink_input *i, const char *event, pa_proplist *pl) {
1613     playback_stream *s;
1614     pa_tagstruct *t;
1615
1616     pa_sink_input_assert_ref(i);
1617     s = PLAYBACK_STREAM(i->userdata);
1618     playback_stream_assert_ref(s);
1619
1620     if (s->connection->version < 15)
1621       return;
1622
1623     t = pa_tagstruct_new(NULL, 0);
1624     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_EVENT);
1625     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1626     pa_tagstruct_putu32(t, s->index);
1627     pa_tagstruct_puts(t, event);
1628     pa_tagstruct_put_proplist(t, pl);
1629     pa_pstream_send_tagstruct(s->connection->pstream, t);
1630 }
1631
1632 /* Called from main context */
1633 static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspend) {
1634     playback_stream *s;
1635     pa_tagstruct *t;
1636
1637     pa_sink_input_assert_ref(i);
1638     s = PLAYBACK_STREAM(i->userdata);
1639     playback_stream_assert_ref(s);
1640
1641     if (s->connection->version < 12)
1642       return;
1643
1644     t = pa_tagstruct_new(NULL, 0);
1645     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_SUSPENDED);
1646     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1647     pa_tagstruct_putu32(t, s->index);
1648     pa_tagstruct_put_boolean(t, suspend);
1649     pa_pstream_send_tagstruct(s->connection->pstream, t);
1650 }
1651
1652 /* Called from main context */
1653 static void sink_input_moving_cb(pa_sink_input *i, pa_sink *dest) {
1654     playback_stream *s;
1655     pa_tagstruct *t;
1656
1657     pa_sink_input_assert_ref(i);
1658     s = PLAYBACK_STREAM(i->userdata);
1659     playback_stream_assert_ref(s);
1660
1661     if (!dest)
1662         return;
1663
1664     fix_playback_buffer_attr(s);
1665     pa_memblockq_apply_attr(s->memblockq, &s->buffer_attr);
1666     pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
1667
1668     if (s->connection->version < 12)
1669       return;
1670
1671     t = pa_tagstruct_new(NULL, 0);
1672     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_MOVED);
1673     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1674     pa_tagstruct_putu32(t, s->index);
1675     pa_tagstruct_putu32(t, dest->index);
1676     pa_tagstruct_puts(t, dest->name);
1677     pa_tagstruct_put_boolean(t, pa_sink_get_state(dest) == PA_SINK_SUSPENDED);
1678
1679     if (s->connection->version >= 13) {
1680         pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
1681         pa_tagstruct_putu32(t, s->buffer_attr.tlength);
1682         pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
1683         pa_tagstruct_putu32(t, s->buffer_attr.minreq);
1684         pa_tagstruct_put_usec(t, s->configured_sink_latency);
1685     }
1686
1687     pa_pstream_send_tagstruct(s->connection->pstream, t);
1688 }
1689
1690 /*** source_output callbacks ***/
1691
1692 /* Called from thread context */
1693 static int source_output_process_msg(pa_msgobject *_o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
1694     pa_source_output *o = PA_SOURCE_OUTPUT(_o);
1695     record_stream *s;
1696
1697     pa_source_output_assert_ref(o);
1698     s = RECORD_STREAM(o->userdata);
1699     record_stream_assert_ref(s);
1700
1701     switch (code) {
1702         case SOURCE_OUTPUT_MESSAGE_UPDATE_LATENCY:
1703             /* Atomically get a snapshot of all timing parameters... */
1704             s->current_monitor_latency = o->source->monitor_of ? pa_sink_get_latency_within_thread(o->source->monitor_of) : 0;
1705             s->current_source_latency = pa_source_get_latency_within_thread(o->source);
1706             s->on_the_fly_snapshot = pa_atomic_load(&s->on_the_fly);
1707             return 0;
1708     }
1709
1710     return pa_source_output_process_msg(_o, code, userdata, offset, chunk);
1711 }
1712
1713 /* Called from thread context */
1714 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
1715     record_stream *s;
1716
1717     pa_source_output_assert_ref(o);
1718     s = RECORD_STREAM(o->userdata);
1719     record_stream_assert_ref(s);
1720     pa_assert(chunk);
1721
1722     pa_atomic_add(&s->on_the_fly, chunk->length);
1723     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), RECORD_STREAM_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
1724 }
1725
1726 static void source_output_kill_cb(pa_source_output *o) {
1727     record_stream *s;
1728
1729     pa_source_output_assert_ref(o);
1730     s = RECORD_STREAM(o->userdata);
1731     record_stream_assert_ref(s);
1732
1733     record_stream_send_killed(s);
1734     record_stream_unlink(s);
1735 }
1736
1737 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
1738     record_stream *s;
1739
1740     pa_source_output_assert_ref(o);
1741     s = RECORD_STREAM(o->userdata);
1742     record_stream_assert_ref(s);
1743
1744     /*pa_log("get_latency: %u", pa_memblockq_get_length(s->memblockq));*/
1745
1746     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &o->sample_spec);
1747 }
1748
1749 /* Called from main context */
1750 static void source_output_send_event_cb(pa_source_output *o, const char *event, pa_proplist *pl) {
1751     record_stream *s;
1752     pa_tagstruct *t;
1753
1754     pa_source_output_assert_ref(o);
1755     s = RECORD_STREAM(o->userdata);
1756     record_stream_assert_ref(s);
1757
1758     if (s->connection->version < 15)
1759       return;
1760
1761     t = pa_tagstruct_new(NULL, 0);
1762     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_EVENT);
1763     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1764     pa_tagstruct_putu32(t, s->index);
1765     pa_tagstruct_puts(t, event);
1766     pa_tagstruct_put_proplist(t, pl);
1767     pa_pstream_send_tagstruct(s->connection->pstream, t);
1768 }
1769
1770 /* Called from main context */
1771 static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend) {
1772     record_stream *s;
1773     pa_tagstruct *t;
1774
1775     pa_source_output_assert_ref(o);
1776     s = RECORD_STREAM(o->userdata);
1777     record_stream_assert_ref(s);
1778
1779     if (s->connection->version < 12)
1780       return;
1781
1782     t = pa_tagstruct_new(NULL, 0);
1783     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_SUSPENDED);
1784     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1785     pa_tagstruct_putu32(t, s->index);
1786     pa_tagstruct_put_boolean(t, suspend);
1787     pa_pstream_send_tagstruct(s->connection->pstream, t);
1788 }
1789
1790 /* Called from main context */
1791 static void source_output_moving_cb(pa_source_output *o, pa_source *dest) {
1792     record_stream *s;
1793     pa_tagstruct *t;
1794
1795     pa_source_output_assert_ref(o);
1796     s = RECORD_STREAM(o->userdata);
1797     record_stream_assert_ref(s);
1798
1799     if (!dest)
1800         return;
1801
1802     fix_record_buffer_attr_pre(s);
1803     pa_memblockq_set_maxlength(s->memblockq, s->buffer_attr.maxlength);
1804     pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
1805     fix_record_buffer_attr_post(s);
1806
1807     if (s->connection->version < 12)
1808       return;
1809
1810     t = pa_tagstruct_new(NULL, 0);
1811     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_MOVED);
1812     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1813     pa_tagstruct_putu32(t, s->index);
1814     pa_tagstruct_putu32(t, dest->index);
1815     pa_tagstruct_puts(t, dest->name);
1816     pa_tagstruct_put_boolean(t, pa_source_get_state(dest) == PA_SOURCE_SUSPENDED);
1817
1818     if (s->connection->version >= 13) {
1819         pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
1820         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
1821         pa_tagstruct_put_usec(t, s->configured_source_latency);
1822     }
1823
1824     pa_pstream_send_tagstruct(s->connection->pstream, t);
1825 }
1826
1827 /*** pdispatch callbacks ***/
1828
1829 static void protocol_error(pa_native_connection *c) {
1830     pa_log("protocol error, kicking client");
1831     native_connection_unlink(c);
1832 }
1833
1834 #define CHECK_VALIDITY(pstream, expression, tag, error) do { \
1835 if (!(expression)) { \
1836     pa_pstream_send_error((pstream), (tag), (error)); \
1837     return; \
1838 } \
1839 } while(0);
1840
1841 static pa_tagstruct *reply_new(uint32_t tag) {
1842     pa_tagstruct *reply;
1843
1844     reply = pa_tagstruct_new(NULL, 0);
1845     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1846     pa_tagstruct_putu32(reply, tag);
1847     return reply;
1848 }
1849
1850 static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1851     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
1852     playback_stream *s;
1853     uint32_t sink_index, syncid, missing;
1854     pa_buffer_attr attr;
1855     const char *name = NULL, *sink_name;
1856     pa_sample_spec ss;
1857     pa_channel_map map;
1858     pa_tagstruct *reply;
1859     pa_sink *sink = NULL;
1860     pa_cvolume volume;
1861     pa_bool_t
1862         corked = FALSE,
1863         no_remap = FALSE,
1864         no_remix = FALSE,
1865         fix_format = FALSE,
1866         fix_rate = FALSE,
1867         fix_channels = FALSE,
1868         no_move = FALSE,
1869         variable_rate = FALSE,
1870         muted = FALSE,
1871         adjust_latency = FALSE,
1872         early_requests = FALSE,
1873         dont_inhibit_auto_suspend = FALSE,
1874         muted_set = FALSE,
1875         fail_on_suspend = FALSE,
1876         relative_volume = FALSE,
1877         passthrough = FALSE;
1878
1879     pa_sink_input_flags_t flags = 0;
1880     pa_proplist *p;
1881     pa_bool_t volume_set = TRUE;
1882     int ret = PA_ERR_INVALID;
1883
1884     pa_native_connection_assert_ref(c);
1885     pa_assert(t);
1886     memset(&attr, 0, sizeof(attr));
1887
1888     if ((c->version < 13 && (pa_tagstruct_gets(t, &name) < 0 || !name)) ||
1889         pa_tagstruct_get(
1890                 t,
1891                 PA_TAG_SAMPLE_SPEC, &ss,
1892                 PA_TAG_CHANNEL_MAP, &map,
1893                 PA_TAG_U32, &sink_index,
1894                 PA_TAG_STRING, &sink_name,
1895                 PA_TAG_U32, &attr.maxlength,
1896                 PA_TAG_BOOLEAN, &corked,
1897                 PA_TAG_U32, &attr.tlength,
1898                 PA_TAG_U32, &attr.prebuf,
1899                 PA_TAG_U32, &attr.minreq,
1900                 PA_TAG_U32, &syncid,
1901                 PA_TAG_CVOLUME, &volume,
1902                 PA_TAG_INVALID) < 0) {
1903
1904         protocol_error(c);
1905         return;
1906     }
1907
1908     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1909     CHECK_VALIDITY(c->pstream, !sink_name || pa_namereg_is_valid_name_or_wildcard(sink_name, PA_NAMEREG_SINK), tag, PA_ERR_INVALID);
1910     CHECK_VALIDITY(c->pstream, sink_index == PA_INVALID_INDEX || !sink_name, tag, PA_ERR_INVALID);
1911     CHECK_VALIDITY(c->pstream, !sink_name || sink_index == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
1912     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1913     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1914     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
1915     CHECK_VALIDITY(c->pstream, map.channels == ss.channels && volume.channels == ss.channels, tag, PA_ERR_INVALID);
1916
1917     p = pa_proplist_new();
1918
1919     if (name)
1920         pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
1921
1922     if (c->version >= 12)  {
1923         /* Since 0.9.8 the user can ask for a couple of additional flags */
1924
1925         if (pa_tagstruct_get_boolean(t, &no_remap) < 0 ||
1926             pa_tagstruct_get_boolean(t, &no_remix) < 0 ||
1927             pa_tagstruct_get_boolean(t, &fix_format) < 0 ||
1928             pa_tagstruct_get_boolean(t, &fix_rate) < 0 ||
1929             pa_tagstruct_get_boolean(t, &fix_channels) < 0 ||
1930             pa_tagstruct_get_boolean(t, &no_move) < 0 ||
1931             pa_tagstruct_get_boolean(t, &variable_rate) < 0) {
1932
1933             protocol_error(c);
1934             pa_proplist_free(p);
1935             return;
1936         }
1937     }
1938
1939     if (c->version >= 13) {
1940
1941         if (pa_tagstruct_get_boolean(t, &muted) < 0 ||
1942             pa_tagstruct_get_boolean(t, &adjust_latency) < 0 ||
1943             pa_tagstruct_get_proplist(t, p) < 0) {
1944             protocol_error(c);
1945             pa_proplist_free(p);
1946             return;
1947         }
1948     }
1949
1950     if (c->version >= 14) {
1951
1952         if (pa_tagstruct_get_boolean(t, &volume_set) < 0 ||
1953             pa_tagstruct_get_boolean(t, &early_requests) < 0) {
1954             protocol_error(c);
1955             pa_proplist_free(p);
1956             return;
1957         }
1958     }
1959
1960     if (c->version >= 15) {
1961
1962         if (pa_tagstruct_get_boolean(t, &muted_set) < 0 ||
1963             pa_tagstruct_get_boolean(t, &dont_inhibit_auto_suspend) < 0 ||
1964             pa_tagstruct_get_boolean(t, &fail_on_suspend) < 0) {
1965             protocol_error(c);
1966             pa_proplist_free(p);
1967             return;
1968         }
1969     }
1970
1971     if (c->version >= 17) {
1972
1973         if (pa_tagstruct_get_boolean(t, &relative_volume) < 0) {
1974             protocol_error(c);
1975             pa_proplist_free(p);
1976             return;
1977         }
1978     }
1979
1980     if (c->version >= 18) {
1981
1982         if (pa_tagstruct_get_boolean(t, &passthrough) < 0 ) {
1983             protocol_error(c);
1984             pa_proplist_free(p);
1985             return;
1986         }
1987     }
1988
1989     if (!pa_tagstruct_eof(t)) {
1990         protocol_error(c);
1991         pa_proplist_free(p);
1992         return;
1993     }
1994
1995     if (sink_index != PA_INVALID_INDEX) {
1996
1997         if (!(sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index))) {
1998             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1999             pa_proplist_free(p);
2000             return;
2001         }
2002
2003     } else if (sink_name) {
2004
2005         if (!(sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK))) {
2006             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2007             pa_proplist_free(p);
2008             return;
2009         }
2010     }
2011
2012     flags =
2013         (corked ?  PA_SINK_INPUT_START_CORKED : 0) |
2014         (no_remap ?  PA_SINK_INPUT_NO_REMAP : 0) |
2015         (no_remix ?  PA_SINK_INPUT_NO_REMIX : 0) |
2016         (fix_format ?  PA_SINK_INPUT_FIX_FORMAT : 0) |
2017         (fix_rate ?  PA_SINK_INPUT_FIX_RATE : 0) |
2018         (fix_channels ?  PA_SINK_INPUT_FIX_CHANNELS : 0) |
2019         (no_move ?  PA_SINK_INPUT_DONT_MOVE : 0) |
2020         (variable_rate ?  PA_SINK_INPUT_VARIABLE_RATE : 0) |
2021         (dont_inhibit_auto_suspend ? PA_SINK_INPUT_DONT_INHIBIT_AUTO_SUSPEND : 0) |
2022         (fail_on_suspend ? PA_SINK_INPUT_NO_CREATE_ON_SUSPEND|PA_SINK_INPUT_KILL_ON_SUSPEND : 0) |
2023         (passthrough ? PA_SINK_INPUT_PASSTHROUGH : 0);
2024
2025     /* Only since protocol version 15 there's a seperate muted_set
2026      * flag. For older versions we synthesize it here */
2027     muted_set = muted_set || muted;
2028
2029     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);
2030     pa_proplist_free(p);
2031
2032     CHECK_VALIDITY(c->pstream, s, tag, ret);
2033
2034     reply = reply_new(tag);
2035     pa_tagstruct_putu32(reply, s->index);
2036     pa_assert(s->sink_input);
2037     pa_tagstruct_putu32(reply, s->sink_input->index);
2038     pa_tagstruct_putu32(reply, missing);
2039
2040 /*     pa_log("initial request is %u", missing); */
2041
2042     if (c->version >= 9) {
2043         /* Since 0.9.0 we support sending the buffer metrics back to the client */
2044
2045         pa_tagstruct_putu32(reply, (uint32_t) s->buffer_attr.maxlength);
2046         pa_tagstruct_putu32(reply, (uint32_t) s->buffer_attr.tlength);
2047         pa_tagstruct_putu32(reply, (uint32_t) s->buffer_attr.prebuf);
2048         pa_tagstruct_putu32(reply, (uint32_t) s->buffer_attr.minreq);
2049     }
2050
2051     if (c->version >= 12) {
2052         /* Since 0.9.8 we support sending the chosen sample
2053          * spec/channel map/device/suspend status back to the
2054          * client */
2055
2056         pa_tagstruct_put_sample_spec(reply, &ss);
2057         pa_tagstruct_put_channel_map(reply, &map);
2058
2059         pa_tagstruct_putu32(reply, s->sink_input->sink->index);
2060         pa_tagstruct_puts(reply, s->sink_input->sink->name);
2061
2062         pa_tagstruct_put_boolean(reply, pa_sink_get_state(s->sink_input->sink) == PA_SINK_SUSPENDED);
2063     }
2064
2065     if (c->version >= 13)
2066         pa_tagstruct_put_usec(reply, s->configured_sink_latency);
2067
2068     pa_pstream_send_tagstruct(c->pstream, reply);
2069 }
2070
2071 static void command_delete_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2072     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2073     uint32_t channel;
2074
2075     pa_native_connection_assert_ref(c);
2076     pa_assert(t);
2077
2078     if (pa_tagstruct_getu32(t, &channel) < 0 ||
2079         !pa_tagstruct_eof(t)) {
2080         protocol_error(c);
2081         return;
2082     }
2083
2084     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2085
2086     switch (command) {
2087
2088         case PA_COMMAND_DELETE_PLAYBACK_STREAM: {
2089             playback_stream *s;
2090             if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !playback_stream_isinstance(s)) {
2091                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
2092                 return;
2093             }
2094
2095             playback_stream_unlink(s);
2096             break;
2097         }
2098
2099         case PA_COMMAND_DELETE_RECORD_STREAM: {
2100             record_stream *s;
2101             if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
2102                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
2103                 return;
2104             }
2105
2106             record_stream_unlink(s);
2107             break;
2108         }
2109
2110         case PA_COMMAND_DELETE_UPLOAD_STREAM: {
2111             upload_stream *s;
2112
2113             if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !upload_stream_isinstance(s)) {
2114                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
2115                 return;
2116             }
2117
2118             upload_stream_unlink(s);
2119             break;
2120         }
2121
2122         default:
2123             pa_assert_not_reached();
2124     }
2125
2126     pa_pstream_send_simple_ack(c->pstream, tag);
2127 }
2128
2129 static void command_create_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2130     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2131     record_stream *s;
2132     pa_buffer_attr attr;
2133     uint32_t source_index;
2134     const char *name = NULL, *source_name;
2135     pa_sample_spec ss;
2136     pa_channel_map map;
2137     pa_tagstruct *reply;
2138     pa_source *source = NULL;
2139     pa_bool_t
2140         corked = FALSE,
2141         no_remap = FALSE,
2142         no_remix = FALSE,
2143         fix_format = FALSE,
2144         fix_rate = FALSE,
2145         fix_channels = FALSE,
2146         no_move = FALSE,
2147         variable_rate = FALSE,
2148         adjust_latency = FALSE,
2149         peak_detect = FALSE,
2150         early_requests = FALSE,
2151         dont_inhibit_auto_suspend = FALSE,
2152         fail_on_suspend = FALSE;
2153     pa_source_output_flags_t flags = 0;
2154     pa_proplist *p;
2155     uint32_t direct_on_input_idx = PA_INVALID_INDEX;
2156     pa_sink_input *direct_on_input = NULL;
2157     int ret = PA_ERR_INVALID;
2158
2159     pa_native_connection_assert_ref(c);
2160     pa_assert(t);
2161
2162     memset(&attr, 0, sizeof(attr));
2163
2164     if ((c->version < 13 && (pa_tagstruct_gets(t, &name) < 0 || !name)) ||
2165         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
2166         pa_tagstruct_get_channel_map(t, &map) < 0 ||
2167         pa_tagstruct_getu32(t, &source_index) < 0 ||
2168         pa_tagstruct_gets(t, &source_name) < 0 ||
2169         pa_tagstruct_getu32(t, &attr.maxlength) < 0 ||
2170         pa_tagstruct_get_boolean(t, &corked) < 0 ||
2171         pa_tagstruct_getu32(t, &attr.fragsize) < 0) {
2172         protocol_error(c);
2173         return;
2174     }
2175
2176     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2177     CHECK_VALIDITY(c->pstream, !source_name || pa_namereg_is_valid_name_or_wildcard(source_name, PA_NAMEREG_SOURCE), tag, PA_ERR_INVALID);
2178     CHECK_VALIDITY(c->pstream, source_index == PA_INVALID_INDEX || !source_name, tag, PA_ERR_INVALID);
2179     CHECK_VALIDITY(c->pstream, !source_name || source_index == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2180     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
2181     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
2182     CHECK_VALIDITY(c->pstream, map.channels == ss.channels, tag, PA_ERR_INVALID);
2183
2184     p = pa_proplist_new();
2185
2186     if (name)
2187         pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
2188
2189     if (c->version >= 12)  {
2190         /* Since 0.9.8 the user can ask for a couple of additional flags */
2191
2192         if (pa_tagstruct_get_boolean(t, &no_remap) < 0 ||
2193             pa_tagstruct_get_boolean(t, &no_remix) < 0 ||
2194             pa_tagstruct_get_boolean(t, &fix_format) < 0 ||
2195             pa_tagstruct_get_boolean(t, &fix_rate) < 0 ||
2196             pa_tagstruct_get_boolean(t, &fix_channels) < 0 ||
2197             pa_tagstruct_get_boolean(t, &no_move) < 0 ||
2198             pa_tagstruct_get_boolean(t, &variable_rate) < 0) {
2199
2200             protocol_error(c);
2201             pa_proplist_free(p);
2202             return;
2203         }
2204     }
2205
2206     if (c->version >= 13) {
2207
2208         if (pa_tagstruct_get_boolean(t, &peak_detect) < 0 ||
2209             pa_tagstruct_get_boolean(t, &adjust_latency) < 0 ||
2210             pa_tagstruct_get_proplist(t, p) < 0 ||
2211             pa_tagstruct_getu32(t, &direct_on_input_idx) < 0) {
2212             protocol_error(c);
2213             pa_proplist_free(p);
2214             return;
2215         }
2216     }
2217
2218     if (c->version >= 14) {
2219
2220         if (pa_tagstruct_get_boolean(t, &early_requests) < 0) {
2221             protocol_error(c);
2222             pa_proplist_free(p);
2223             return;
2224         }
2225     }
2226
2227     if (c->version >= 15) {
2228
2229         if (pa_tagstruct_get_boolean(t, &dont_inhibit_auto_suspend) < 0 ||
2230             pa_tagstruct_get_boolean(t, &fail_on_suspend) < 0) {
2231             protocol_error(c);
2232             pa_proplist_free(p);
2233             return;
2234         }
2235     }
2236
2237     if (!pa_tagstruct_eof(t)) {
2238         protocol_error(c);
2239         pa_proplist_free(p);
2240         return;
2241     }
2242
2243     if (source_index != PA_INVALID_INDEX) {
2244
2245         if (!(source = pa_idxset_get_by_index(c->protocol->core->sources, source_index))) {
2246             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2247             pa_proplist_free(p);
2248             return;
2249         }
2250
2251     } else if (source_name) {
2252
2253         if (!(source = pa_namereg_get(c->protocol->core, source_name, PA_NAMEREG_SOURCE))) {
2254             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2255             pa_proplist_free(p);
2256             return;
2257         }
2258     }
2259
2260     if (direct_on_input_idx != PA_INVALID_INDEX) {
2261
2262         if (!(direct_on_input = pa_idxset_get_by_index(c->protocol->core->sink_inputs, direct_on_input_idx))) {
2263             pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2264             pa_proplist_free(p);
2265             return;
2266         }
2267     }
2268
2269     flags =
2270         (corked ?  PA_SOURCE_OUTPUT_START_CORKED : 0) |
2271         (no_remap ?  PA_SOURCE_OUTPUT_NO_REMAP : 0) |
2272         (no_remix ?  PA_SOURCE_OUTPUT_NO_REMIX : 0) |
2273         (fix_format ?  PA_SOURCE_OUTPUT_FIX_FORMAT : 0) |
2274         (fix_rate ?  PA_SOURCE_OUTPUT_FIX_RATE : 0) |
2275         (fix_channels ?  PA_SOURCE_OUTPUT_FIX_CHANNELS : 0) |
2276         (no_move ?  PA_SOURCE_OUTPUT_DONT_MOVE : 0) |
2277         (variable_rate ?  PA_SOURCE_OUTPUT_VARIABLE_RATE : 0) |
2278         (dont_inhibit_auto_suspend ? PA_SOURCE_OUTPUT_DONT_INHIBIT_AUTO_SUSPEND : 0) |
2279         (fail_on_suspend ? PA_SOURCE_OUTPUT_NO_CREATE_ON_SUSPEND|PA_SOURCE_OUTPUT_KILL_ON_SUSPEND : 0);
2280
2281     s = record_stream_new(c, source, &ss, &map, peak_detect, &attr, flags, p, adjust_latency, direct_on_input, early_requests, &ret);
2282     pa_proplist_free(p);
2283
2284     CHECK_VALIDITY(c->pstream, s, tag, ret);
2285
2286     reply = reply_new(tag);
2287     pa_tagstruct_putu32(reply, s->index);
2288     pa_assert(s->source_output);
2289     pa_tagstruct_putu32(reply, s->source_output->index);
2290
2291     if (c->version >= 9) {
2292         /* Since 0.9 we support sending the buffer metrics back to the client */
2293
2294         pa_tagstruct_putu32(reply, (uint32_t) s->buffer_attr.maxlength);
2295         pa_tagstruct_putu32(reply, (uint32_t) s->buffer_attr.fragsize);
2296     }
2297
2298     if (c->version >= 12) {
2299         /* Since 0.9.8 we support sending the chosen sample
2300          * spec/channel map/device/suspend status back to the
2301          * client */
2302
2303         pa_tagstruct_put_sample_spec(reply, &ss);
2304         pa_tagstruct_put_channel_map(reply, &map);
2305
2306         pa_tagstruct_putu32(reply, s->source_output->source->index);
2307         pa_tagstruct_puts(reply, s->source_output->source->name);
2308
2309         pa_tagstruct_put_boolean(reply, pa_source_get_state(s->source_output->source) == PA_SOURCE_SUSPENDED);
2310     }
2311
2312     if (c->version >= 13)
2313         pa_tagstruct_put_usec(reply, s->configured_source_latency);
2314
2315     pa_pstream_send_tagstruct(c->pstream, reply);
2316 }
2317
2318 static void command_exit(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2319     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2320     int ret;
2321
2322     pa_native_connection_assert_ref(c);
2323     pa_assert(t);
2324
2325     if (!pa_tagstruct_eof(t)) {
2326         protocol_error(c);
2327         return;
2328     }
2329
2330     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2331     ret = pa_core_exit(c->protocol->core, FALSE, 0);
2332     CHECK_VALIDITY(c->pstream, ret >= 0, tag, PA_ERR_ACCESS);
2333
2334     pa_log_debug("Client %s asks us to terminate.", pa_strnull(pa_proplist_gets(c->client->proplist, PA_PROP_APPLICATION_PROCESS_BINARY)));
2335
2336     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
2337 }
2338
2339 static void command_auth(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2340     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2341     const void*cookie;
2342     pa_tagstruct *reply;
2343     pa_bool_t shm_on_remote = FALSE, do_shm;
2344
2345     pa_native_connection_assert_ref(c);
2346     pa_assert(t);
2347
2348     if (pa_tagstruct_getu32(t, &c->version) < 0 ||
2349         pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
2350         !pa_tagstruct_eof(t)) {
2351         protocol_error(c);
2352         return;
2353     }
2354
2355     /* Minimum supported version */
2356     if (c->version < 8) {
2357         pa_pstream_send_error(c->pstream, tag, PA_ERR_VERSION);
2358         return;
2359     }
2360
2361     /* Starting with protocol version 13 the MSB of the version tag
2362        reflects if shm is available for this pa_native_connection or
2363        not. */
2364     if (c->version >= 13) {
2365         shm_on_remote = !!(c->version & 0x80000000U);
2366         c->version &= 0x7FFFFFFFU;
2367     }
2368
2369     pa_log_debug("Protocol version: remote %u, local %u", c->version, PA_PROTOCOL_VERSION);
2370
2371     pa_proplist_setf(c->client->proplist, "native-protocol.version", "%u", c->version);
2372
2373     if (!c->authorized) {
2374         pa_bool_t success = FALSE;
2375
2376 #ifdef HAVE_CREDS
2377         const pa_creds *creds;
2378
2379         if ((creds = pa_pdispatch_creds(pd))) {
2380             if (creds->uid == getuid())
2381                 success = TRUE;
2382             else if (c->options->auth_group) {
2383                 int r;
2384                 gid_t gid;
2385
2386                 if ((gid = pa_get_gid_of_group(c->options->auth_group)) == (gid_t) -1)
2387                     pa_log_warn("Failed to get GID of group '%s'", c->options->auth_group);
2388                 else if (gid == creds->gid)
2389                     success = TRUE;
2390
2391                 if (!success) {
2392                     if ((r = pa_uid_in_group(creds->uid, c->options->auth_group)) < 0)
2393                         pa_log_warn("Failed to check group membership.");
2394                     else if (r > 0)
2395                         success = TRUE;
2396                 }
2397             }
2398
2399             pa_log_info("Got credentials: uid=%lu gid=%lu success=%i",
2400                         (unsigned long) creds->uid,
2401                         (unsigned long) creds->gid,
2402                         (int) success);
2403         }
2404 #endif
2405
2406         if (!success && c->options->auth_cookie) {
2407             const uint8_t *ac;
2408
2409             if ((ac = pa_auth_cookie_read(c->options->auth_cookie, PA_NATIVE_COOKIE_LENGTH)))
2410                 if (memcmp(ac, cookie, PA_NATIVE_COOKIE_LENGTH) == 0)
2411                     success = TRUE;
2412         }
2413
2414         if (!success) {
2415             pa_log_warn("Denied access to client with invalid authorization data.");
2416             pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS);
2417             return;
2418         }
2419
2420         c->authorized = TRUE;
2421         if (c->auth_timeout_event) {
2422             c->protocol->core->mainloop->time_free(c->auth_timeout_event);
2423             c->auth_timeout_event = NULL;
2424         }
2425     }
2426
2427     /* Enable shared memory support if possible */
2428     do_shm =
2429         pa_mempool_is_shared(c->protocol->core->mempool) &&
2430         c->is_local;
2431
2432     pa_log_debug("SHM possible: %s", pa_yes_no(do_shm));
2433
2434     if (do_shm)
2435         if (c->version < 10 || (c->version >= 13 && !shm_on_remote))
2436             do_shm = FALSE;
2437
2438 #ifdef HAVE_CREDS
2439     if (do_shm) {
2440         /* Only enable SHM if both sides are owned by the same
2441          * user. This is a security measure because otherwise data
2442          * private to the user might leak. */
2443
2444         const pa_creds *creds;
2445         if (!(creds = pa_pdispatch_creds(pd)) || getuid() != creds->uid)
2446             do_shm = FALSE;
2447     }
2448 #endif
2449
2450     pa_log_debug("Negotiated SHM: %s", pa_yes_no(do_shm));
2451     pa_pstream_enable_shm(c->pstream, do_shm);
2452
2453     reply = reply_new(tag);
2454     pa_tagstruct_putu32(reply, PA_PROTOCOL_VERSION | (do_shm ? 0x80000000 : 0));
2455
2456 #ifdef HAVE_CREDS
2457 {
2458     /* SHM support is only enabled after both sides made sure they are the same user. */
2459
2460     pa_creds ucred;
2461
2462     ucred.uid = getuid();
2463     ucred.gid = getgid();
2464
2465     pa_pstream_send_tagstruct_with_creds(c->pstream, reply, &ucred);
2466 }
2467 #else
2468     pa_pstream_send_tagstruct(c->pstream, reply);
2469 #endif
2470 }
2471
2472 static void command_set_client_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2473     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2474     const char *name = NULL;
2475     pa_proplist *p;
2476     pa_tagstruct *reply;
2477
2478     pa_native_connection_assert_ref(c);
2479     pa_assert(t);
2480
2481     p = pa_proplist_new();
2482
2483     if ((c->version < 13 && pa_tagstruct_gets(t, &name) < 0) ||
2484         (c->version >= 13 && pa_tagstruct_get_proplist(t, p) < 0) ||
2485         !pa_tagstruct_eof(t)) {
2486
2487         protocol_error(c);
2488         pa_proplist_free(p);
2489         return;
2490     }
2491
2492     if (name)
2493         if (pa_proplist_sets(p, PA_PROP_APPLICATION_NAME, name) < 0) {
2494             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
2495             pa_proplist_free(p);
2496             return;
2497         }
2498
2499     pa_client_update_proplist(c->client, PA_UPDATE_REPLACE, p);
2500     pa_proplist_free(p);
2501
2502     reply = reply_new(tag);
2503
2504     if (c->version >= 13)
2505         pa_tagstruct_putu32(reply, c->client->index);
2506
2507     pa_pstream_send_tagstruct(c->pstream, reply);
2508 }
2509
2510 static void command_lookup(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2511     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2512     const char *name;
2513     uint32_t idx = PA_IDXSET_INVALID;
2514
2515     pa_native_connection_assert_ref(c);
2516     pa_assert(t);
2517
2518     if (pa_tagstruct_gets(t, &name) < 0 ||
2519         !pa_tagstruct_eof(t)) {
2520         protocol_error(c);
2521         return;
2522     }
2523
2524     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2525     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);
2526
2527     if (command == PA_COMMAND_LOOKUP_SINK) {
2528         pa_sink *sink;
2529         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK)))
2530             idx = sink->index;
2531     } else {
2532         pa_source *source;
2533         pa_assert(command == PA_COMMAND_LOOKUP_SOURCE);
2534         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE)))
2535             idx = source->index;
2536     }
2537
2538     if (idx == PA_IDXSET_INVALID)
2539         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2540     else {
2541         pa_tagstruct *reply;
2542         reply = reply_new(tag);
2543         pa_tagstruct_putu32(reply, idx);
2544         pa_pstream_send_tagstruct(c->pstream, reply);
2545     }
2546 }
2547
2548 static void command_drain_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2549     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2550     uint32_t idx;
2551     playback_stream *s;
2552
2553     pa_native_connection_assert_ref(c);
2554     pa_assert(t);
2555
2556     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2557         !pa_tagstruct_eof(t)) {
2558         protocol_error(c);
2559         return;
2560     }
2561
2562     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2563     s = pa_idxset_get_by_index(c->output_streams, idx);
2564     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2565     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2566
2567     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);
2568 }
2569
2570 static void command_stat(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2571     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2572     pa_tagstruct *reply;
2573     const pa_mempool_stat *stat;
2574
2575     pa_native_connection_assert_ref(c);
2576     pa_assert(t);
2577
2578     if (!pa_tagstruct_eof(t)) {
2579         protocol_error(c);
2580         return;
2581     }
2582
2583     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2584
2585     stat = pa_mempool_get_stat(c->protocol->core->mempool);
2586
2587     reply = reply_new(tag);
2588     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->n_allocated));
2589     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->allocated_size));
2590     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->n_accumulated));
2591     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->accumulated_size));
2592     pa_tagstruct_putu32(reply, (uint32_t) pa_scache_total_size(c->protocol->core));
2593     pa_pstream_send_tagstruct(c->pstream, reply);
2594 }
2595
2596 static void command_get_playback_latency(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2597     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2598     pa_tagstruct *reply;
2599     playback_stream *s;
2600     struct timeval tv, now;
2601     uint32_t idx;
2602
2603     pa_native_connection_assert_ref(c);
2604     pa_assert(t);
2605
2606     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2607         pa_tagstruct_get_timeval(t, &tv) < 0 ||
2608         !pa_tagstruct_eof(t)) {
2609         protocol_error(c);
2610         return;
2611     }
2612
2613     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2614     s = pa_idxset_get_by_index(c->output_streams, idx);
2615     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2616     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2617
2618     /* Get an atomic snapshot of all timing parameters */
2619     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);
2620
2621     reply = reply_new(tag);
2622     pa_tagstruct_put_usec(reply,
2623                           s->current_sink_latency +
2624                           pa_bytes_to_usec(s->render_memblockq_length, &s->sink_input->sink->sample_spec));
2625     pa_tagstruct_put_usec(reply, 0);
2626     pa_tagstruct_put_boolean(reply,
2627                              s->playing_for > 0 &&
2628                              pa_sink_get_state(s->sink_input->sink) == PA_SINK_RUNNING &&
2629                              pa_sink_input_get_state(s->sink_input) == PA_SINK_INPUT_RUNNING);
2630     pa_tagstruct_put_timeval(reply, &tv);
2631     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
2632     pa_tagstruct_puts64(reply, s->write_index);
2633     pa_tagstruct_puts64(reply, s->read_index);
2634
2635     if (c->version >= 13) {
2636         pa_tagstruct_putu64(reply, s->underrun_for);
2637         pa_tagstruct_putu64(reply, s->playing_for);
2638     }
2639
2640     pa_pstream_send_tagstruct(c->pstream, reply);
2641 }
2642
2643 static void command_get_record_latency(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2644     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2645     pa_tagstruct *reply;
2646     record_stream *s;
2647     struct timeval tv, now;
2648     uint32_t idx;
2649
2650     pa_native_connection_assert_ref(c);
2651     pa_assert(t);
2652
2653     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2654         pa_tagstruct_get_timeval(t, &tv) < 0 ||
2655         !pa_tagstruct_eof(t)) {
2656         protocol_error(c);
2657         return;
2658     }
2659
2660     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2661     s = pa_idxset_get_by_index(c->record_streams, idx);
2662     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2663
2664     /* Get an atomic snapshot of all timing parameters */
2665     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);
2666
2667     reply = reply_new(tag);
2668     pa_tagstruct_put_usec(reply, s->current_monitor_latency);
2669     pa_tagstruct_put_usec(reply,
2670                           s->current_source_latency +
2671                           pa_bytes_to_usec(s->on_the_fly_snapshot, &s->source_output->source->sample_spec));
2672     pa_tagstruct_put_boolean(reply,
2673                              pa_source_get_state(s->source_output->source) == PA_SOURCE_RUNNING &&
2674                              pa_source_output_get_state(s->source_output) == PA_SOURCE_OUTPUT_RUNNING);
2675     pa_tagstruct_put_timeval(reply, &tv);
2676     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
2677     pa_tagstruct_puts64(reply, pa_memblockq_get_write_index(s->memblockq));
2678     pa_tagstruct_puts64(reply, pa_memblockq_get_read_index(s->memblockq));
2679     pa_pstream_send_tagstruct(c->pstream, reply);
2680 }
2681
2682 static void command_create_upload_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2683     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2684     upload_stream *s;
2685     uint32_t length;
2686     const char *name = NULL;
2687     pa_sample_spec ss;
2688     pa_channel_map map;
2689     pa_tagstruct *reply;
2690     pa_proplist *p;
2691
2692     pa_native_connection_assert_ref(c);
2693     pa_assert(t);
2694
2695     if (pa_tagstruct_gets(t, &name) < 0 ||
2696         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
2697         pa_tagstruct_get_channel_map(t, &map) < 0 ||
2698         pa_tagstruct_getu32(t, &length) < 0) {
2699         protocol_error(c);
2700         return;
2701     }
2702
2703     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2704     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
2705     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
2706     CHECK_VALIDITY(c->pstream, map.channels == ss.channels, tag, PA_ERR_INVALID);
2707     CHECK_VALIDITY(c->pstream, (length % pa_frame_size(&ss)) == 0 && length > 0, tag, PA_ERR_INVALID);
2708     CHECK_VALIDITY(c->pstream, length <= PA_SCACHE_ENTRY_SIZE_MAX, tag, PA_ERR_TOOLARGE);
2709
2710     p = pa_proplist_new();
2711
2712     if ((c->version >= 13 && pa_tagstruct_get_proplist(t, p) < 0) ||
2713         !pa_tagstruct_eof(t)) {
2714
2715         protocol_error(c);
2716         pa_proplist_free(p);
2717         return;
2718     }
2719
2720     if (c->version < 13)
2721         pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
2722     else if (!name)
2723         if (!(name = pa_proplist_gets(p, PA_PROP_EVENT_ID)))
2724             name = pa_proplist_gets(p, PA_PROP_MEDIA_NAME);
2725
2726     if (!name || !pa_namereg_is_valid_name(name)) {
2727         pa_proplist_free(p);
2728         CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_INVALID);
2729     }
2730
2731     s = upload_stream_new(c, &ss, &map, name, length, p);
2732     pa_proplist_free(p);
2733
2734     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
2735
2736     reply = reply_new(tag);
2737     pa_tagstruct_putu32(reply, s->index);
2738     pa_tagstruct_putu32(reply, length);
2739     pa_pstream_send_tagstruct(c->pstream, reply);
2740 }
2741
2742 static void command_finish_upload_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2743     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2744     uint32_t channel;
2745     upload_stream *s;
2746     uint32_t idx;
2747
2748     pa_native_connection_assert_ref(c);
2749     pa_assert(t);
2750
2751     if (pa_tagstruct_getu32(t, &channel) < 0 ||
2752         !pa_tagstruct_eof(t)) {
2753         protocol_error(c);
2754         return;
2755     }
2756
2757     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2758
2759     s = pa_idxset_get_by_index(c->output_streams, channel);
2760     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2761     CHECK_VALIDITY(c->pstream, upload_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2762
2763     if (!s->memchunk.memblock)
2764         pa_pstream_send_error(c->pstream, tag, PA_ERR_TOOLARGE);
2765     else if (pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->channel_map, &s->memchunk, s->proplist, &idx) < 0)
2766         pa_pstream_send_error(c->pstream, tag, PA_ERR_INTERNAL);
2767     else
2768         pa_pstream_send_simple_ack(c->pstream, tag);
2769
2770     upload_stream_unlink(s);
2771 }
2772
2773 static void command_play_sample(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2774     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2775     uint32_t sink_index;
2776     pa_volume_t volume;
2777     pa_sink *sink;
2778     const char *name, *sink_name;
2779     uint32_t idx;
2780     pa_proplist *p;
2781     pa_tagstruct *reply;
2782
2783     pa_native_connection_assert_ref(c);
2784     pa_assert(t);
2785
2786     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2787
2788     if (pa_tagstruct_getu32(t, &sink_index) < 0 ||
2789         pa_tagstruct_gets(t, &sink_name) < 0 ||
2790         pa_tagstruct_getu32(t, &volume) < 0 ||
2791         pa_tagstruct_gets(t, &name) < 0) {
2792         protocol_error(c);
2793         return;
2794     }
2795
2796     CHECK_VALIDITY(c->pstream, !sink_name || pa_namereg_is_valid_name_or_wildcard(sink_name, PA_NAMEREG_SINK), tag, PA_ERR_INVALID);
2797     CHECK_VALIDITY(c->pstream, sink_index == PA_INVALID_INDEX || !sink_name, tag, PA_ERR_INVALID);
2798     CHECK_VALIDITY(c->pstream, !sink_name || sink_index == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2799     CHECK_VALIDITY(c->pstream, name && pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
2800
2801     if (sink_index != PA_INVALID_INDEX)
2802         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
2803     else
2804         sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK);
2805
2806     CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
2807
2808     p = pa_proplist_new();
2809
2810     if ((c->version >= 13 && pa_tagstruct_get_proplist(t, p) < 0) ||
2811         !pa_tagstruct_eof(t)) {
2812         protocol_error(c);
2813         pa_proplist_free(p);
2814         return;
2815     }
2816
2817     pa_proplist_update(p, PA_UPDATE_MERGE, c->client->proplist);
2818
2819     if (pa_scache_play_item(c->protocol->core, name, sink, volume, p, &idx) < 0) {
2820         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2821         pa_proplist_free(p);
2822         return;
2823     }
2824
2825     pa_proplist_free(p);
2826
2827     reply = reply_new(tag);
2828
2829     if (c->version >= 13)
2830         pa_tagstruct_putu32(reply, idx);
2831
2832     pa_pstream_send_tagstruct(c->pstream, reply);
2833 }
2834
2835 static void command_remove_sample(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2836     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
2837     const char *name;
2838
2839     pa_native_connection_assert_ref(c);
2840     pa_assert(t);
2841
2842     if (pa_tagstruct_gets(t, &name) < 0 ||
2843         !pa_tagstruct_eof(t)) {
2844         protocol_error(c);
2845         return;
2846     }
2847
2848     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2849     CHECK_VALIDITY(c->pstream, name && pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
2850
2851     if (pa_scache_remove_item(c->protocol->core, name) < 0) {
2852         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2853         return;
2854     }
2855
2856     pa_pstream_send_simple_ack(c->pstream, tag);
2857 }
2858
2859 static void fixup_sample_spec(pa_native_connection *c, pa_sample_spec *fixed, const pa_sample_spec *original) {
2860     pa_assert(c);
2861     pa_assert(fixed);
2862     pa_assert(original);
2863
2864     *fixed = *original;
2865
2866     if (c->version < 12) {
2867         /* Before protocol version 12 we didn't support S32 samples,
2868          * so we need to lie about this to the client */
2869
2870         if (fixed->format == PA_SAMPLE_S32LE)
2871             fixed->format = PA_SAMPLE_FLOAT32LE;
2872         if (fixed->format == PA_SAMPLE_S32BE)
2873             fixed->format = PA_SAMPLE_FLOAT32BE;
2874     }
2875
2876     if (c->version < 15) {
2877         if (fixed->format == PA_SAMPLE_S24LE || fixed->format == PA_SAMPLE_S24_32LE)
2878             fixed->format = PA_SAMPLE_FLOAT32LE;
2879         if (fixed->format == PA_SAMPLE_S24BE || fixed->format == PA_SAMPLE_S24_32BE)
2880             fixed->format = PA_SAMPLE_FLOAT32BE;
2881     }
2882 }
2883
2884 static void sink_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_sink *sink) {
2885     pa_sample_spec fixed_ss;
2886
2887     pa_assert(t);
2888     pa_sink_assert_ref(sink);
2889
2890     fixup_sample_spec(c, &fixed_ss, &sink->sample_spec);
2891
2892     pa_tagstruct_put(
2893         t,
2894         PA_TAG_U32, sink->index,
2895         PA_TAG_STRING, sink->name,
2896         PA_TAG_STRING, pa_strnull(pa_proplist_gets(sink->proplist, PA_PROP_DEVICE_DESCRIPTION)),
2897         PA_TAG_SAMPLE_SPEC, &fixed_ss,
2898         PA_TAG_CHANNEL_MAP, &sink->channel_map,
2899         PA_TAG_U32, sink->module ? sink->module->index : PA_INVALID_INDEX,
2900         PA_TAG_CVOLUME, pa_sink_get_volume(sink, FALSE),
2901         PA_TAG_BOOLEAN, pa_sink_get_mute(sink, FALSE),
2902         PA_TAG_U32, sink->monitor_source ? sink->monitor_source->index : PA_INVALID_INDEX,
2903         PA_TAG_STRING, sink->monitor_source ? sink->monitor_source->name : NULL,
2904         PA_TAG_USEC, pa_sink_get_latency(sink),
2905         PA_TAG_STRING, sink->driver,
2906         PA_TAG_U32, sink->flags,
2907         PA_TAG_INVALID);
2908
2909     if (c->version >= 13) {
2910         pa_tagstruct_put_proplist(t, sink->proplist);
2911         pa_tagstruct_put_usec(t, pa_sink_get_requested_latency(sink));
2912     }
2913
2914     if (c->version >= 15) {
2915         pa_tagstruct_put_volume(t, sink->base_volume);
2916         if (PA_UNLIKELY(pa_sink_get_state(sink) == PA_SINK_INVALID_STATE))
2917             pa_log_error("Internal sink state is invalid.");
2918         pa_tagstruct_putu32(t, pa_sink_get_state(sink));
2919         pa_tagstruct_putu32(t, sink->n_volume_steps);
2920         pa_tagstruct_putu32(t, sink->card ? sink->card->index : PA_INVALID_INDEX);
2921     }
2922
2923     if (c->version >= 16) {
2924         pa_tagstruct_putu32(t, sink->ports ? pa_hashmap_size(sink->ports) : 0);
2925
2926         if (sink->ports) {
2927             void *state;
2928             pa_device_port *p;
2929
2930             PA_HASHMAP_FOREACH(p, sink->ports, state) {
2931                 pa_tagstruct_puts(t, p->name);
2932                 pa_tagstruct_puts(t, p->description);
2933                 pa_tagstruct_putu32(t, p->priority);
2934             }
2935         }
2936
2937         pa_tagstruct_puts(t, sink->active_port ? sink->active_port->name : NULL);
2938     }
2939 }
2940
2941 static void source_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_source *source) {
2942     pa_sample_spec fixed_ss;
2943
2944     pa_assert(t);
2945     pa_source_assert_ref(source);
2946
2947     fixup_sample_spec(c, &fixed_ss, &source->sample_spec);
2948
2949     pa_tagstruct_put(
2950         t,
2951         PA_TAG_U32, source->index,
2952         PA_TAG_STRING, source->name,
2953         PA_TAG_STRING, pa_strnull(pa_proplist_gets(source->proplist, PA_PROP_DEVICE_DESCRIPTION)),
2954         PA_TAG_SAMPLE_SPEC, &fixed_ss,
2955         PA_TAG_CHANNEL_MAP, &source->channel_map,
2956         PA_TAG_U32, source->module ? source->module->index : PA_INVALID_INDEX,
2957         PA_TAG_CVOLUME, pa_source_get_volume(source, FALSE),
2958         PA_TAG_BOOLEAN, pa_source_get_mute(source, FALSE),
2959         PA_TAG_U32, source->monitor_of ? source->monitor_of->index : PA_INVALID_INDEX,
2960         PA_TAG_STRING, source->monitor_of ? source->monitor_of->name : NULL,
2961         PA_TAG_USEC, pa_source_get_latency(source),
2962         PA_TAG_STRING, source->driver,
2963         PA_TAG_U32, source->flags,
2964         PA_TAG_INVALID);
2965
2966     if (c->version >= 13) {
2967         pa_tagstruct_put_proplist(t, source->proplist);
2968         pa_tagstruct_put_usec(t, pa_source_get_requested_latency(source));
2969     }
2970
2971     if (c->version >= 15) {
2972         pa_tagstruct_put_volume(t, source->base_volume);
2973         if (PA_UNLIKELY(pa_source_get_state(source) == PA_SOURCE_INVALID_STATE))
2974             pa_log_error("Internal source state is invalid.");
2975         pa_tagstruct_putu32(t, pa_source_get_state(source));
2976         pa_tagstruct_putu32(t, source->n_volume_steps);
2977         pa_tagstruct_putu32(t, source->card ? source->card->index : PA_INVALID_INDEX);
2978     }
2979
2980     if (c->version >= 16) {
2981
2982         pa_tagstruct_putu32(t, source->ports ? pa_hashmap_size(source->ports) : 0);
2983
2984         if (source->ports) {
2985             void *state;
2986             pa_device_port *p;
2987
2988             PA_HASHMAP_FOREACH(p, source->ports, state) {
2989                 pa_tagstruct_puts(t, p->name);
2990                 pa_tagstruct_puts(t, p->description);
2991                 pa_tagstruct_putu32(t, p->priority);
2992             }
2993         }
2994
2995         pa_tagstruct_puts(t, source->active_port ? source->active_port->name : NULL);
2996     }
2997 }
2998
2999 static void client_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_client *client) {
3000     pa_assert(t);
3001     pa_assert(client);
3002
3003     pa_tagstruct_putu32(t, client->index);
3004     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(client->proplist, PA_PROP_APPLICATION_NAME)));
3005     pa_tagstruct_putu32(t, client->module ? client->module->index : PA_INVALID_INDEX);
3006     pa_tagstruct_puts(t, client->driver);
3007
3008     if (c->version >= 13)
3009         pa_tagstruct_put_proplist(t, client->proplist);
3010 }
3011
3012 static void card_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_card *card) {
3013     void *state = NULL;
3014     pa_card_profile *p;
3015
3016     pa_assert(t);
3017     pa_assert(card);
3018
3019     pa_tagstruct_putu32(t, card->index);
3020     pa_tagstruct_puts(t, card->name);
3021     pa_tagstruct_putu32(t, card->module ? card->module->index : PA_INVALID_INDEX);
3022     pa_tagstruct_puts(t, card->driver);
3023
3024     pa_tagstruct_putu32(t, card->profiles ? pa_hashmap_size(card->profiles) : 0);
3025
3026     if (card->profiles) {
3027         while ((p = pa_hashmap_iterate(card->profiles, &state, NULL))) {
3028             pa_tagstruct_puts(t, p->name);
3029             pa_tagstruct_puts(t, p->description);
3030             pa_tagstruct_putu32(t, p->n_sinks);
3031             pa_tagstruct_putu32(t, p->n_sources);
3032             pa_tagstruct_putu32(t, p->priority);
3033         }
3034     }
3035
3036     pa_tagstruct_puts(t, card->active_profile ? card->active_profile->name : NULL);
3037     pa_tagstruct_put_proplist(t, card->proplist);
3038 }
3039
3040 static void module_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_module *module) {
3041     pa_assert(t);
3042     pa_assert(module);
3043
3044     pa_tagstruct_putu32(t, module->index);
3045     pa_tagstruct_puts(t, module->name);
3046     pa_tagstruct_puts(t, module->argument);
3047     pa_tagstruct_putu32(t, (uint32_t) pa_module_get_n_used(module));
3048
3049     if (c->version < 15)
3050         pa_tagstruct_put_boolean(t, FALSE); /* autoload is obsolete */
3051
3052     if (c->version >= 15)
3053         pa_tagstruct_put_proplist(t, module->proplist);
3054 }
3055
3056 static void sink_input_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_sink_input *s) {
3057     pa_sample_spec fixed_ss;
3058     pa_usec_t sink_latency;
3059     pa_cvolume v;
3060
3061     pa_assert(t);
3062     pa_sink_input_assert_ref(s);
3063
3064     fixup_sample_spec(c, &fixed_ss, &s->sample_spec);
3065
3066     pa_tagstruct_putu32(t, s->index);
3067     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME)));
3068     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
3069     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
3070     pa_tagstruct_putu32(t, s->sink->index);
3071     pa_tagstruct_put_sample_spec(t, &fixed_ss);
3072     pa_tagstruct_put_channel_map(t, &s->channel_map);
3073     pa_tagstruct_put_cvolume(t, pa_sink_input_get_volume(s, &v, TRUE));
3074     pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s, &sink_latency));
3075     pa_tagstruct_put_usec(t, sink_latency);
3076     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
3077     pa_tagstruct_puts(t, s->driver);
3078     if (c->version >= 11)
3079         pa_tagstruct_put_boolean(t, pa_sink_input_get_mute(s));
3080     if (c->version >= 13)
3081         pa_tagstruct_put_proplist(t, s->proplist);
3082     if (c->version >= 19)
3083         pa_tagstruct_put_boolean(t, (pa_sink_input_get_state(s) == PA_SINK_INPUT_CORKED));
3084 }
3085
3086 static void source_output_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_source_output *s) {
3087     pa_sample_spec fixed_ss;
3088     pa_usec_t source_latency;
3089
3090     pa_assert(t);
3091     pa_source_output_assert_ref(s);
3092
3093     fixup_sample_spec(c, &fixed_ss, &s->sample_spec);
3094
3095     pa_tagstruct_putu32(t, s->index);
3096     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME)));
3097     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
3098     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
3099     pa_tagstruct_putu32(t, s->source->index);
3100     pa_tagstruct_put_sample_spec(t, &fixed_ss);
3101     pa_tagstruct_put_channel_map(t, &s->channel_map);
3102     pa_tagstruct_put_usec(t, pa_source_output_get_latency(s, &source_latency));
3103     pa_tagstruct_put_usec(t, source_latency);
3104     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_source_output_get_resample_method(s)));
3105     pa_tagstruct_puts(t, s->driver);
3106     if (c->version >= 13)
3107         pa_tagstruct_put_proplist(t, s->proplist);
3108     if (c->version >= 19)
3109         pa_tagstruct_put_boolean(t, (pa_source_output_get_state(s) == PA_SOURCE_OUTPUT_CORKED));
3110 }
3111
3112 static void scache_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_scache_entry *e) {
3113     pa_sample_spec fixed_ss;
3114     pa_cvolume v;
3115
3116     pa_assert(t);
3117     pa_assert(e);
3118
3119     if (e->memchunk.memblock)
3120         fixup_sample_spec(c, &fixed_ss, &e->sample_spec);
3121     else
3122         memset(&fixed_ss, 0, sizeof(fixed_ss));
3123
3124     pa_tagstruct_putu32(t, e->index);
3125     pa_tagstruct_puts(t, e->name);
3126
3127     if (e->volume_is_set)
3128         v = e->volume;
3129     else
3130         pa_cvolume_init(&v);
3131
3132     pa_tagstruct_put_cvolume(t, &v);
3133     pa_tagstruct_put_usec(t, e->memchunk.memblock ? pa_bytes_to_usec(e->memchunk.length, &e->sample_spec) : 0);
3134     pa_tagstruct_put_sample_spec(t, &fixed_ss);
3135     pa_tagstruct_put_channel_map(t, &e->channel_map);
3136     pa_tagstruct_putu32(t, (uint32_t) e->memchunk.length);
3137     pa_tagstruct_put_boolean(t, e->lazy);
3138     pa_tagstruct_puts(t, e->filename);
3139
3140     if (c->version >= 13)
3141         pa_tagstruct_put_proplist(t, e->proplist);
3142 }
3143
3144 static void command_get_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3145     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3146     uint32_t idx;
3147     pa_sink *sink = NULL;
3148     pa_source *source = NULL;
3149     pa_client *client = NULL;
3150     pa_card *card = NULL;
3151     pa_module *module = NULL;
3152     pa_sink_input *si = NULL;
3153     pa_source_output *so = NULL;
3154     pa_scache_entry *sce = NULL;
3155     const char *name = NULL;
3156     pa_tagstruct *reply;
3157
3158     pa_native_connection_assert_ref(c);
3159     pa_assert(t);
3160
3161     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3162         (command != PA_COMMAND_GET_CLIENT_INFO &&
3163          command != PA_COMMAND_GET_MODULE_INFO &&
3164          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
3165          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
3166          pa_tagstruct_gets(t, &name) < 0) ||
3167         !pa_tagstruct_eof(t)) {
3168         protocol_error(c);
3169         return;
3170     }
3171
3172     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3173     CHECK_VALIDITY(c->pstream, !name ||
3174                    (command == PA_COMMAND_GET_SINK_INFO &&
3175                     pa_namereg_is_valid_name_or_wildcard(name, PA_NAMEREG_SINK)) ||
3176                    (command == PA_COMMAND_GET_SOURCE_INFO &&
3177                     pa_namereg_is_valid_name_or_wildcard(name, PA_NAMEREG_SOURCE)) ||
3178                    pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
3179     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
3180     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
3181     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3182
3183     if (command == PA_COMMAND_GET_SINK_INFO) {
3184         if (idx != PA_INVALID_INDEX)
3185             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3186         else
3187             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
3188     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
3189         if (idx != PA_INVALID_INDEX)
3190             source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3191         else
3192             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
3193     } else if (command == PA_COMMAND_GET_CARD_INFO) {
3194         if (idx != PA_INVALID_INDEX)
3195             card = pa_idxset_get_by_index(c->protocol->core->cards, idx);
3196         else
3197             card = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_CARD);
3198     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
3199         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
3200     else if (command == PA_COMMAND_GET_MODULE_INFO)
3201         module = pa_idxset_get_by_index(c->protocol->core->modules, idx);
3202     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
3203         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3204     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
3205         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
3206     else {
3207         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO);
3208         if (idx != PA_INVALID_INDEX)
3209             sce = pa_idxset_get_by_index(c->protocol->core->scache, idx);
3210         else
3211             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE);
3212     }
3213
3214     if (!sink && !source && !client && !card && !module && !si && !so && !sce) {
3215         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
3216         return;
3217     }
3218
3219     reply = reply_new(tag);
3220     if (sink)
3221         sink_fill_tagstruct(c, reply, sink);
3222     else if (source)
3223         source_fill_tagstruct(c, reply, source);
3224     else if (client)
3225         client_fill_tagstruct(c, reply, client);
3226     else if (card)
3227         card_fill_tagstruct(c, reply, card);
3228     else if (module)
3229         module_fill_tagstruct(c, reply, module);
3230     else if (si)
3231         sink_input_fill_tagstruct(c, reply, si);
3232     else if (so)
3233         source_output_fill_tagstruct(c, reply, so);
3234     else
3235         scache_fill_tagstruct(c, reply, sce);
3236     pa_pstream_send_tagstruct(c->pstream, reply);
3237 }
3238
3239 static void command_get_info_list(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3240     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3241     pa_idxset *i;
3242     uint32_t idx;
3243     void *p;
3244     pa_tagstruct *reply;
3245
3246     pa_native_connection_assert_ref(c);
3247     pa_assert(t);
3248
3249     if (!pa_tagstruct_eof(t)) {
3250         protocol_error(c);
3251         return;
3252     }
3253
3254     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3255
3256     reply = reply_new(tag);
3257
3258     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
3259         i = c->protocol->core->sinks;
3260     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
3261         i = c->protocol->core->sources;
3262     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
3263         i = c->protocol->core->clients;
3264     else if (command == PA_COMMAND_GET_CARD_INFO_LIST)
3265         i = c->protocol->core->cards;
3266     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
3267         i = c->protocol->core->modules;
3268     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
3269         i = c->protocol->core->sink_inputs;
3270     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
3271         i = c->protocol->core->source_outputs;
3272     else {
3273         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
3274         i = c->protocol->core->scache;
3275     }
3276
3277     if (i) {
3278         for (p = pa_idxset_first(i, &idx); p; p = pa_idxset_next(i, &idx)) {
3279             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
3280                 sink_fill_tagstruct(c, reply, p);
3281             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
3282                 source_fill_tagstruct(c, reply, p);
3283             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
3284                 client_fill_tagstruct(c, reply, p);
3285             else if (command == PA_COMMAND_GET_CARD_INFO_LIST)
3286                 card_fill_tagstruct(c, reply, p);
3287             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
3288                 module_fill_tagstruct(c, reply, p);
3289             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
3290                 sink_input_fill_tagstruct(c, reply, p);
3291             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
3292                 source_output_fill_tagstruct(c, reply, p);
3293             else {
3294                 pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
3295                 scache_fill_tagstruct(c, reply, p);
3296             }
3297         }
3298     }
3299
3300     pa_pstream_send_tagstruct(c->pstream, reply);
3301 }
3302
3303 static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3304     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3305     pa_tagstruct *reply;
3306     pa_sink *def_sink;
3307     pa_source *def_source;
3308     pa_sample_spec fixed_ss;
3309     char *h, *u;
3310
3311     pa_native_connection_assert_ref(c);
3312     pa_assert(t);
3313
3314     if (!pa_tagstruct_eof(t)) {
3315         protocol_error(c);
3316         return;
3317     }
3318
3319     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3320
3321     reply = reply_new(tag);
3322     pa_tagstruct_puts(reply, PACKAGE_NAME);
3323     pa_tagstruct_puts(reply, PACKAGE_VERSION);
3324
3325     u = pa_get_user_name_malloc();
3326     pa_tagstruct_puts(reply, u);
3327     pa_xfree(u);
3328
3329     h = pa_get_host_name_malloc();
3330     pa_tagstruct_puts(reply, h);
3331     pa_xfree(h);
3332
3333     fixup_sample_spec(c, &fixed_ss, &c->protocol->core->default_sample_spec);
3334     pa_tagstruct_put_sample_spec(reply, &fixed_ss);
3335
3336     def_sink = pa_namereg_get_default_sink(c->protocol->core);
3337     pa_tagstruct_puts(reply, def_sink ? def_sink->name : NULL);
3338     def_source = pa_namereg_get_default_source(c->protocol->core);
3339     pa_tagstruct_puts(reply, def_source ? def_source->name : NULL);
3340
3341     pa_tagstruct_putu32(reply, c->protocol->core->cookie);
3342
3343     if (c->version >= 15)
3344         pa_tagstruct_put_channel_map(reply, &c->protocol->core->default_channel_map);
3345
3346     pa_pstream_send_tagstruct(c->pstream, reply);
3347 }
3348
3349 static void subscription_cb(pa_core *core, pa_subscription_event_type_t e, uint32_t idx, void *userdata) {
3350     pa_tagstruct *t;
3351     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3352
3353     pa_native_connection_assert_ref(c);
3354
3355     t = pa_tagstruct_new(NULL, 0);
3356     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
3357     pa_tagstruct_putu32(t, (uint32_t) -1);
3358     pa_tagstruct_putu32(t, e);
3359     pa_tagstruct_putu32(t, idx);
3360     pa_pstream_send_tagstruct(c->pstream, t);
3361 }
3362
3363 static void command_subscribe(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3364     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3365     pa_subscription_mask_t m;
3366
3367     pa_native_connection_assert_ref(c);
3368     pa_assert(t);
3369
3370     if (pa_tagstruct_getu32(t, &m) < 0 ||
3371         !pa_tagstruct_eof(t)) {
3372         protocol_error(c);
3373         return;
3374     }
3375
3376     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3377     CHECK_VALIDITY(c->pstream, (m & ~PA_SUBSCRIPTION_MASK_ALL) == 0, tag, PA_ERR_INVALID);
3378
3379     if (c->subscription)
3380         pa_subscription_free(c->subscription);
3381
3382     if (m != 0) {
3383         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
3384         pa_assert(c->subscription);
3385     } else
3386         c->subscription = NULL;
3387
3388     pa_pstream_send_simple_ack(c->pstream, tag);
3389 }
3390
3391 static void command_set_volume(
3392         pa_pdispatch *pd,
3393         uint32_t command,
3394         uint32_t tag,
3395         pa_tagstruct *t,
3396         void *userdata) {
3397
3398     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3399     uint32_t idx;
3400     pa_cvolume volume;
3401     pa_sink *sink = NULL;
3402     pa_source *source = NULL;
3403     pa_sink_input *si = NULL;
3404     const char *name = NULL;
3405     const char *client_name;
3406
3407     pa_native_connection_assert_ref(c);
3408     pa_assert(t);
3409
3410     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3411         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
3412         (command == PA_COMMAND_SET_SOURCE_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
3413         pa_tagstruct_get_cvolume(t, &volume) ||
3414         !pa_tagstruct_eof(t)) {
3415         protocol_error(c);
3416         return;
3417     }
3418
3419     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3420     CHECK_VALIDITY(c->pstream, !name || pa_namereg_is_valid_name_or_wildcard(name, command == PA_COMMAND_SET_SINK_VOLUME ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE), tag, PA_ERR_INVALID);
3421     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
3422     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
3423     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3424     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
3425
3426     switch (command) {
3427
3428         case PA_COMMAND_SET_SINK_VOLUME:
3429             if (idx != PA_INVALID_INDEX)
3430                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3431             else
3432                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
3433             break;
3434
3435         case PA_COMMAND_SET_SOURCE_VOLUME:
3436             if (idx != PA_INVALID_INDEX)
3437                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3438             else
3439                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
3440             break;
3441
3442         case PA_COMMAND_SET_SINK_INPUT_VOLUME:
3443             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3444             break;
3445
3446         default:
3447             pa_assert_not_reached();
3448     }
3449
3450     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
3451
3452     client_name = pa_strnull(pa_proplist_gets(c->client->proplist, PA_PROP_APPLICATION_PROCESS_BINARY));
3453
3454     if (sink) {
3455         CHECK_VALIDITY(c->pstream, volume.channels == 1 || pa_cvolume_compatible(&volume, &sink->sample_spec), tag, PA_ERR_INVALID);
3456
3457         pa_log_debug("Client %s changes volume of sink %s.", client_name, sink->name);
3458         pa_sink_set_volume(sink, &volume, TRUE, TRUE);
3459     } else if (source) {
3460         CHECK_VALIDITY(c->pstream, volume.channels == 1 || pa_cvolume_compatible(&volume, &source->sample_spec), tag, PA_ERR_INVALID);
3461
3462         pa_log_debug("Client %s changes volume of source %s.", client_name, source->name);
3463         pa_source_set_volume(source, &volume, TRUE);
3464     } else if (si) {
3465         CHECK_VALIDITY(c->pstream, volume.channels == 1 || pa_cvolume_compatible(&volume, &si->sample_spec), tag, PA_ERR_INVALID);
3466
3467         pa_log_debug("Client %s changes volume of sink input %s.",
3468                      client_name,
3469                      pa_strnull(pa_proplist_gets(si->proplist, PA_PROP_MEDIA_NAME)));
3470         pa_sink_input_set_volume(si, &volume, TRUE, TRUE);
3471     }
3472
3473     pa_pstream_send_simple_ack(c->pstream, tag);
3474 }
3475
3476 static void command_set_mute(
3477         pa_pdispatch *pd,
3478         uint32_t command,
3479         uint32_t tag,
3480         pa_tagstruct *t,
3481         void *userdata) {
3482
3483     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3484     uint32_t idx;
3485     pa_bool_t mute;
3486     pa_sink *sink = NULL;
3487     pa_source *source = NULL;
3488     pa_sink_input *si = NULL;
3489     const char *name = NULL, *client_name;
3490
3491     pa_native_connection_assert_ref(c);
3492     pa_assert(t);
3493
3494     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3495         (command == PA_COMMAND_SET_SINK_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
3496         (command == PA_COMMAND_SET_SOURCE_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
3497         pa_tagstruct_get_boolean(t, &mute) ||
3498         !pa_tagstruct_eof(t)) {
3499         protocol_error(c);
3500         return;
3501     }
3502
3503     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3504     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);
3505     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
3506     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
3507     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3508
3509     switch (command) {
3510
3511         case PA_COMMAND_SET_SINK_MUTE:
3512             if (idx != PA_INVALID_INDEX)
3513                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3514             else
3515                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
3516
3517             break;
3518
3519         case PA_COMMAND_SET_SOURCE_MUTE:
3520             if (idx != PA_INVALID_INDEX)
3521                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3522             else
3523                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
3524
3525             break;
3526
3527         case PA_COMMAND_SET_SINK_INPUT_MUTE:
3528             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3529             break;
3530
3531         default:
3532             pa_assert_not_reached();
3533     }
3534
3535     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
3536
3537     client_name = pa_strnull(pa_proplist_gets(c->client->proplist, PA_PROP_APPLICATION_PROCESS_BINARY));
3538
3539     if (sink) {
3540         pa_log_debug("Client %s changes mute of sink %s.", client_name, sink->name);
3541         pa_sink_set_mute(sink, mute, TRUE);
3542     } else if (source) {
3543         pa_log_debug("Client %s changes mute of source %s.", client_name, source->name);
3544         pa_source_set_mute(source, mute, TRUE);
3545     } else if (si) {
3546         pa_log_debug("Client %s changes mute of sink input %s.",
3547                      client_name,
3548                      pa_strnull(pa_proplist_gets(si->proplist, PA_PROP_MEDIA_NAME)));
3549         pa_sink_input_set_mute(si, mute, TRUE);
3550     }
3551
3552     pa_pstream_send_simple_ack(c->pstream, tag);
3553 }
3554
3555 static void command_cork_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3556     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3557     uint32_t idx;
3558     pa_bool_t b;
3559     playback_stream *s;
3560
3561     pa_native_connection_assert_ref(c);
3562     pa_assert(t);
3563
3564     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3565         pa_tagstruct_get_boolean(t, &b) < 0 ||
3566         !pa_tagstruct_eof(t)) {
3567         protocol_error(c);
3568         return;
3569     }
3570
3571     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3572     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3573     s = pa_idxset_get_by_index(c->output_streams, idx);
3574     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3575     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3576
3577     pa_sink_input_cork(s->sink_input, b);
3578
3579     if (b)
3580         s->is_underrun = TRUE;
3581
3582     pa_pstream_send_simple_ack(c->pstream, tag);
3583 }
3584
3585 static void command_trigger_or_flush_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3586     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3587     uint32_t idx;
3588     playback_stream *s;
3589
3590     pa_native_connection_assert_ref(c);
3591     pa_assert(t);
3592
3593     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3594         !pa_tagstruct_eof(t)) {
3595         protocol_error(c);
3596         return;
3597     }
3598
3599     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3600     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3601     s = pa_idxset_get_by_index(c->output_streams, idx);
3602     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3603     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3604
3605     switch (command) {
3606         case PA_COMMAND_FLUSH_PLAYBACK_STREAM:
3607             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_FLUSH, NULL, 0, NULL);
3608             break;
3609
3610         case PA_COMMAND_PREBUF_PLAYBACK_STREAM:
3611             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_PREBUF_FORCE, NULL, 0, NULL);
3612             break;
3613
3614         case PA_COMMAND_TRIGGER_PLAYBACK_STREAM:
3615             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_TRIGGER, NULL, 0, NULL);
3616             break;
3617
3618         default:
3619             pa_assert_not_reached();
3620     }
3621
3622     pa_pstream_send_simple_ack(c->pstream, tag);
3623 }
3624
3625 static void command_cork_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3626     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3627     uint32_t idx;
3628     record_stream *s;
3629     pa_bool_t b;
3630
3631     pa_native_connection_assert_ref(c);
3632     pa_assert(t);
3633
3634     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3635         pa_tagstruct_get_boolean(t, &b) < 0 ||
3636         !pa_tagstruct_eof(t)) {
3637         protocol_error(c);
3638         return;
3639     }
3640
3641     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3642     s = pa_idxset_get_by_index(c->record_streams, idx);
3643     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3644
3645     pa_source_output_cork(s->source_output, b);
3646     pa_memblockq_prebuf_force(s->memblockq);
3647     pa_pstream_send_simple_ack(c->pstream, tag);
3648 }
3649
3650 static void command_flush_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3651     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3652     uint32_t idx;
3653     record_stream *s;
3654
3655     pa_native_connection_assert_ref(c);
3656     pa_assert(t);
3657
3658     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3659         !pa_tagstruct_eof(t)) {
3660         protocol_error(c);
3661         return;
3662     }
3663
3664     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3665     s = pa_idxset_get_by_index(c->record_streams, idx);
3666     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3667
3668     pa_memblockq_flush_read(s->memblockq);
3669     pa_pstream_send_simple_ack(c->pstream, tag);
3670 }
3671
3672 static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3673     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3674     uint32_t idx;
3675     pa_buffer_attr a;
3676     pa_tagstruct *reply;
3677
3678     pa_native_connection_assert_ref(c);
3679     pa_assert(t);
3680
3681     memset(&a, 0, sizeof(a));
3682
3683     if (pa_tagstruct_getu32(t, &idx) < 0) {
3684         protocol_error(c);
3685         return;
3686     }
3687
3688     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3689
3690     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR) {
3691         playback_stream *s;
3692         pa_bool_t adjust_latency = FALSE, early_requests = FALSE;
3693
3694         s = pa_idxset_get_by_index(c->output_streams, idx);
3695         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3696         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3697
3698         if (pa_tagstruct_get(
3699                     t,
3700                     PA_TAG_U32, &a.maxlength,
3701                     PA_TAG_U32, &a.tlength,
3702                     PA_TAG_U32, &a.prebuf,
3703                     PA_TAG_U32, &a.minreq,
3704                     PA_TAG_INVALID) < 0 ||
3705             (c->version >= 13 && pa_tagstruct_get_boolean(t, &adjust_latency) < 0) ||
3706             (c->version >= 14 && pa_tagstruct_get_boolean(t, &early_requests) < 0) ||
3707             !pa_tagstruct_eof(t)) {
3708             protocol_error(c);
3709             return;
3710         }
3711
3712         s->adjust_latency = adjust_latency;
3713         s->early_requests = early_requests;
3714         s->buffer_attr = a;
3715
3716         fix_playback_buffer_attr(s);
3717         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);
3718
3719         reply = reply_new(tag);
3720         pa_tagstruct_putu32(reply, s->buffer_attr.maxlength);
3721         pa_tagstruct_putu32(reply, s->buffer_attr.tlength);
3722         pa_tagstruct_putu32(reply, s->buffer_attr.prebuf);
3723         pa_tagstruct_putu32(reply, s->buffer_attr.minreq);
3724
3725         if (c->version >= 13)
3726             pa_tagstruct_put_usec(reply, s->configured_sink_latency);
3727
3728     } else {
3729         record_stream *s;
3730         pa_bool_t adjust_latency = FALSE, early_requests = FALSE;
3731         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR);
3732
3733         s = pa_idxset_get_by_index(c->record_streams, idx);
3734         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3735
3736         if (pa_tagstruct_get(
3737                     t,
3738                     PA_TAG_U32, &a.maxlength,
3739                     PA_TAG_U32, &a.fragsize,
3740                     PA_TAG_INVALID) < 0 ||
3741             (c->version >= 13 && pa_tagstruct_get_boolean(t, &adjust_latency) < 0) ||
3742             (c->version >= 14 && pa_tagstruct_get_boolean(t, &early_requests) < 0) ||
3743             !pa_tagstruct_eof(t)) {
3744             protocol_error(c);
3745             return;
3746         }
3747
3748         s->adjust_latency = adjust_latency;
3749         s->early_requests = early_requests;
3750         s->buffer_attr = a;
3751
3752         fix_record_buffer_attr_pre(s);
3753         pa_memblockq_set_maxlength(s->memblockq, s->buffer_attr.maxlength);
3754         pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
3755         fix_record_buffer_attr_post(s);
3756
3757         reply = reply_new(tag);
3758         pa_tagstruct_putu32(reply, s->buffer_attr.maxlength);
3759         pa_tagstruct_putu32(reply, s->buffer_attr.fragsize);
3760
3761         if (c->version >= 13)
3762             pa_tagstruct_put_usec(reply, s->configured_source_latency);
3763     }
3764
3765     pa_pstream_send_tagstruct(c->pstream, reply);
3766 }
3767
3768 static void command_update_stream_sample_rate(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3769     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3770     uint32_t idx;
3771     uint32_t rate;
3772
3773     pa_native_connection_assert_ref(c);
3774     pa_assert(t);
3775
3776     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3777         pa_tagstruct_getu32(t, &rate) < 0 ||
3778         !pa_tagstruct_eof(t)) {
3779         protocol_error(c);
3780         return;
3781     }
3782
3783     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3784     CHECK_VALIDITY(c->pstream, rate > 0 && rate <= PA_RATE_MAX, tag, PA_ERR_INVALID);
3785
3786     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE) {
3787         playback_stream *s;
3788
3789         s = pa_idxset_get_by_index(c->output_streams, idx);
3790         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3791         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3792
3793         pa_sink_input_set_rate(s->sink_input, rate);
3794
3795     } else {
3796         record_stream *s;
3797         pa_assert(command == PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE);
3798
3799         s = pa_idxset_get_by_index(c->record_streams, idx);
3800         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3801
3802         pa_source_output_set_rate(s->source_output, rate);
3803     }
3804
3805     pa_pstream_send_simple_ack(c->pstream, tag);
3806 }
3807
3808 static void command_update_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3809     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3810     uint32_t idx;
3811     uint32_t mode;
3812     pa_proplist *p;
3813
3814     pa_native_connection_assert_ref(c);
3815     pa_assert(t);
3816
3817     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3818
3819     p = pa_proplist_new();
3820
3821     if (command == PA_COMMAND_UPDATE_CLIENT_PROPLIST) {
3822
3823         if (pa_tagstruct_getu32(t, &mode) < 0 ||
3824             pa_tagstruct_get_proplist(t, p) < 0 ||
3825             !pa_tagstruct_eof(t)) {
3826             protocol_error(c);
3827             pa_proplist_free(p);
3828             return;
3829         }
3830
3831     } else {
3832
3833         if (pa_tagstruct_getu32(t, &idx) < 0 ||
3834             pa_tagstruct_getu32(t, &mode) < 0 ||
3835             pa_tagstruct_get_proplist(t, p) < 0 ||
3836             !pa_tagstruct_eof(t)) {
3837             protocol_error(c);
3838             pa_proplist_free(p);
3839             return;
3840         }
3841     }
3842
3843     if (!(mode == PA_UPDATE_SET || mode == PA_UPDATE_MERGE || mode == PA_UPDATE_REPLACE)) {
3844         pa_proplist_free(p);
3845         CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_INVALID);
3846     }
3847
3848     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST) {
3849         playback_stream *s;
3850
3851         s = pa_idxset_get_by_index(c->output_streams, idx);
3852         if (!s || !playback_stream_isinstance(s)) {
3853             pa_proplist_free(p);
3854             CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_NOENTITY);
3855         }
3856         pa_sink_input_update_proplist(s->sink_input, mode, p);
3857
3858     } else if (command == PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST) {
3859         record_stream *s;
3860
3861         if (!(s = pa_idxset_get_by_index(c->record_streams, idx))) {
3862             pa_proplist_free(p);
3863             CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_NOENTITY);
3864         }
3865         pa_source_output_update_proplist(s->source_output, mode, p);
3866
3867     } else {
3868         pa_assert(command == PA_COMMAND_UPDATE_CLIENT_PROPLIST);
3869
3870         pa_client_update_proplist(c->client, mode, p);
3871     }
3872
3873     pa_pstream_send_simple_ack(c->pstream, tag);
3874     pa_proplist_free(p);
3875 }
3876
3877 static void command_remove_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3878     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3879     uint32_t idx;
3880     unsigned changed = 0;
3881     pa_proplist *p;
3882     pa_strlist *l = NULL;
3883
3884     pa_native_connection_assert_ref(c);
3885     pa_assert(t);
3886
3887     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3888
3889     if (command != PA_COMMAND_REMOVE_CLIENT_PROPLIST) {
3890
3891         if (pa_tagstruct_getu32(t, &idx) < 0) {
3892             protocol_error(c);
3893             return;
3894         }
3895     }
3896
3897     if (command == PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST) {
3898         playback_stream *s;
3899
3900         s = pa_idxset_get_by_index(c->output_streams, idx);
3901         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3902         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3903
3904         p = s->sink_input->proplist;
3905
3906     } else if (command == PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST) {
3907         record_stream *s;
3908
3909         s = pa_idxset_get_by_index(c->record_streams, idx);
3910         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3911
3912         p = s->source_output->proplist;
3913     } else {
3914         pa_assert(command == PA_COMMAND_REMOVE_CLIENT_PROPLIST);
3915
3916         p = c->client->proplist;
3917     }
3918
3919     for (;;) {
3920         const char *k;
3921
3922         if (pa_tagstruct_gets(t, &k) < 0) {
3923             protocol_error(c);
3924             pa_strlist_free(l);
3925             return;
3926         }
3927
3928         if (!k)
3929             break;
3930
3931         l = pa_strlist_prepend(l, k);
3932     }
3933
3934     if (!pa_tagstruct_eof(t)) {
3935         protocol_error(c);
3936         pa_strlist_free(l);
3937         return;
3938     }
3939
3940     for (;;) {
3941         char *z;
3942
3943         l = pa_strlist_pop(l, &z);
3944
3945         if (!z)
3946             break;
3947
3948         changed += (unsigned) (pa_proplist_unset(p, z) >= 0);
3949         pa_xfree(z);
3950     }
3951
3952     pa_pstream_send_simple_ack(c->pstream, tag);
3953
3954     if (changed) {
3955         if (command == PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST) {
3956             playback_stream *s;
3957
3958             s = pa_idxset_get_by_index(c->output_streams, idx);
3959             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->sink_input->index);
3960
3961         } else if (command == PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST) {
3962             record_stream *s;
3963
3964             s = pa_idxset_get_by_index(c->record_streams, idx);
3965             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->source_output->index);
3966
3967         } else {
3968             pa_assert(command == PA_COMMAND_REMOVE_CLIENT_PROPLIST);
3969             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_CLIENT|PA_SUBSCRIPTION_EVENT_CHANGE, c->client->index);
3970         }
3971     }
3972 }
3973
3974 static void command_set_default_sink_or_source(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3975     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3976     const char *s;
3977
3978     pa_native_connection_assert_ref(c);
3979     pa_assert(t);
3980
3981     if (pa_tagstruct_gets(t, &s) < 0 ||
3982         !pa_tagstruct_eof(t)) {
3983         protocol_error(c);
3984         return;
3985     }
3986
3987     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3988     CHECK_VALIDITY(c->pstream, !s || pa_namereg_is_valid_name(s), tag, PA_ERR_INVALID);
3989
3990     if (command == PA_COMMAND_SET_DEFAULT_SOURCE) {
3991         pa_source *source;
3992
3993         source = pa_namereg_get(c->protocol->core, s, PA_NAMEREG_SOURCE);
3994         CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
3995
3996         pa_namereg_set_default_source(c->protocol->core, source);
3997     } else {
3998         pa_sink *sink;
3999         pa_assert(command == PA_COMMAND_SET_DEFAULT_SINK);
4000
4001         sink = pa_namereg_get(c->protocol->core, s, PA_NAMEREG_SINK);
4002         CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
4003
4004         pa_namereg_set_default_sink(c->protocol->core, sink);
4005     }
4006
4007     pa_pstream_send_simple_ack(c->pstream, tag);
4008 }
4009
4010 static void command_set_stream_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4011     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4012     uint32_t idx;
4013     const char *name;
4014
4015     pa_native_connection_assert_ref(c);
4016     pa_assert(t);
4017
4018     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4019         pa_tagstruct_gets(t, &name) < 0 ||
4020         !pa_tagstruct_eof(t)) {
4021         protocol_error(c);
4022         return;
4023     }
4024
4025     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4026     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
4027
4028     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
4029         playback_stream *s;
4030
4031         s = pa_idxset_get_by_index(c->output_streams, idx);
4032         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4033         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
4034
4035         pa_sink_input_set_name(s->sink_input, name);
4036
4037     } else {
4038         record_stream *s;
4039         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_NAME);
4040
4041         s = pa_idxset_get_by_index(c->record_streams, idx);
4042         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4043
4044         pa_source_output_set_name(s->source_output, name);
4045     }
4046
4047     pa_pstream_send_simple_ack(c->pstream, tag);
4048 }
4049
4050 static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4051     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4052     uint32_t idx;
4053
4054     pa_native_connection_assert_ref(c);
4055     pa_assert(t);
4056
4057     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4058         !pa_tagstruct_eof(t)) {
4059         protocol_error(c);
4060         return;
4061     }
4062
4063     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4064
4065     if (command == PA_COMMAND_KILL_CLIENT) {
4066         pa_client *client;
4067
4068         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
4069         CHECK_VALIDITY(c->pstream, client, tag, PA_ERR_NOENTITY);
4070
4071         pa_native_connection_ref(c);
4072         pa_client_kill(client);
4073
4074     } else if (command == PA_COMMAND_KILL_SINK_INPUT) {
4075         pa_sink_input *s;
4076
4077         s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
4078         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4079
4080         pa_native_connection_ref(c);
4081         pa_sink_input_kill(s);
4082     } else {
4083         pa_source_output *s;
4084
4085         pa_assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
4086
4087         s = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
4088         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4089
4090         pa_native_connection_ref(c);
4091         pa_source_output_kill(s);
4092     }
4093
4094     pa_pstream_send_simple_ack(c->pstream, tag);
4095     pa_native_connection_unref(c);
4096 }
4097
4098 static void command_load_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4099     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4100     pa_module *m;
4101     const char *name, *argument;
4102     pa_tagstruct *reply;
4103
4104     pa_native_connection_assert_ref(c);
4105     pa_assert(t);
4106
4107     if (pa_tagstruct_gets(t, &name) < 0 ||
4108         pa_tagstruct_gets(t, &argument) < 0 ||
4109         !pa_tagstruct_eof(t)) {
4110         protocol_error(c);
4111         return;
4112     }
4113
4114     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4115     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name) && !strchr(name, '/'), tag, PA_ERR_INVALID);
4116     CHECK_VALIDITY(c->pstream, !argument || pa_utf8_valid(argument), tag, PA_ERR_INVALID);
4117
4118     if (!(m = pa_module_load(c->protocol->core, name, argument))) {
4119         pa_pstream_send_error(c->pstream, tag, PA_ERR_MODINITFAILED);
4120         return;
4121     }
4122
4123     reply = reply_new(tag);
4124     pa_tagstruct_putu32(reply, m->index);
4125     pa_pstream_send_tagstruct(c->pstream, reply);
4126 }
4127
4128 static void command_unload_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4129     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4130     uint32_t idx;
4131     pa_module *m;
4132
4133     pa_native_connection_assert_ref(c);
4134     pa_assert(t);
4135
4136     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4137         !pa_tagstruct_eof(t)) {
4138         protocol_error(c);
4139         return;
4140     }
4141
4142     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4143     m = pa_idxset_get_by_index(c->protocol->core->modules, idx);
4144     CHECK_VALIDITY(c->pstream, m, tag, PA_ERR_NOENTITY);
4145
4146     pa_module_unload_request(m, FALSE);
4147     pa_pstream_send_simple_ack(c->pstream, tag);
4148 }
4149
4150 static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4151     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4152     uint32_t idx = PA_INVALID_INDEX, idx_device = PA_INVALID_INDEX;
4153     const char *name_device = NULL;
4154
4155     pa_native_connection_assert_ref(c);
4156     pa_assert(t);
4157
4158     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4159         pa_tagstruct_getu32(t, &idx_device) < 0 ||
4160         pa_tagstruct_gets(t, &name_device) < 0 ||
4161         !pa_tagstruct_eof(t)) {
4162         protocol_error(c);
4163         return;
4164     }
4165
4166     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4167     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4168
4169     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);
4170     CHECK_VALIDITY(c->pstream, idx_device != PA_INVALID_INDEX || name_device, tag, PA_ERR_INVALID);
4171     CHECK_VALIDITY(c->pstream, idx_device == PA_INVALID_INDEX || !name_device, tag, PA_ERR_INVALID);
4172     CHECK_VALIDITY(c->pstream, !name_device || idx_device == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4173
4174     if (command == PA_COMMAND_MOVE_SINK_INPUT) {
4175         pa_sink_input *si = NULL;
4176         pa_sink *sink = NULL;
4177
4178         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
4179
4180         if (idx_device != PA_INVALID_INDEX)
4181             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx_device);
4182         else
4183             sink = pa_namereg_get(c->protocol->core, name_device, PA_NAMEREG_SINK);
4184
4185         CHECK_VALIDITY(c->pstream, si && sink, tag, PA_ERR_NOENTITY);
4186
4187         if (pa_sink_input_move_to(si, sink, TRUE) < 0) {
4188             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4189             return;
4190         }
4191     } else {
4192         pa_source_output *so = NULL;
4193         pa_source *source;
4194
4195         pa_assert(command == PA_COMMAND_MOVE_SOURCE_OUTPUT);
4196
4197         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
4198
4199         if (idx_device != PA_INVALID_INDEX)
4200             source = pa_idxset_get_by_index(c->protocol->core->sources, idx_device);
4201         else
4202             source = pa_namereg_get(c->protocol->core, name_device, PA_NAMEREG_SOURCE);
4203
4204         CHECK_VALIDITY(c->pstream, so && source, tag, PA_ERR_NOENTITY);
4205
4206         if (pa_source_output_move_to(so, source, TRUE) < 0) {
4207             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4208             return;
4209         }
4210     }
4211
4212     pa_pstream_send_simple_ack(c->pstream, tag);
4213 }
4214
4215 static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4216     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4217     uint32_t idx = PA_INVALID_INDEX;
4218     const char *name = NULL;
4219     pa_bool_t b;
4220
4221     pa_native_connection_assert_ref(c);
4222     pa_assert(t);
4223
4224     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4225         pa_tagstruct_gets(t, &name) < 0 ||
4226         pa_tagstruct_get_boolean(t, &b) < 0 ||
4227         !pa_tagstruct_eof(t)) {
4228         protocol_error(c);
4229         return;
4230     }
4231
4232     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4233     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);
4234     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4235     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4236     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4237
4238     if (command == PA_COMMAND_SUSPEND_SINK) {
4239
4240         if (idx == PA_INVALID_INDEX && name && !*name) {
4241
4242             pa_log_debug("%s all sinks", b ? "Suspending" : "Resuming");
4243
4244             if (pa_sink_suspend_all(c->protocol->core, b, PA_SUSPEND_USER) < 0) {
4245                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4246                 return;
4247             }
4248         } else {
4249             pa_sink *sink = NULL;
4250
4251             if (idx != PA_INVALID_INDEX)
4252                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
4253             else
4254                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
4255
4256             CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
4257
4258             if (pa_sink_suspend(sink, b, PA_SUSPEND_USER) < 0) {
4259                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4260                 return;
4261             }
4262         }
4263     } else {
4264
4265         pa_assert(command == PA_COMMAND_SUSPEND_SOURCE);
4266
4267         if (idx == PA_INVALID_INDEX && name && !*name) {
4268
4269             pa_log_debug("%s all sources", b ? "Suspending" : "Resuming");
4270
4271             if (pa_source_suspend_all(c->protocol->core, b, PA_SUSPEND_USER) < 0) {
4272                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4273                 return;
4274             }
4275
4276         } else {
4277             pa_source *source;
4278
4279             if (idx != PA_INVALID_INDEX)
4280                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
4281             else
4282                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
4283
4284             CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
4285
4286             if (pa_source_suspend(source, b, PA_SUSPEND_USER) < 0) {
4287                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4288                 return;
4289             }
4290         }
4291     }
4292
4293     pa_pstream_send_simple_ack(c->pstream, tag);
4294 }
4295
4296 static void command_extension(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4297     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4298     uint32_t idx = PA_INVALID_INDEX;
4299     const char *name = NULL;
4300     pa_module *m;
4301     pa_native_protocol_ext_cb_t cb;
4302
4303     pa_native_connection_assert_ref(c);
4304     pa_assert(t);
4305
4306     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4307         pa_tagstruct_gets(t, &name) < 0) {
4308         protocol_error(c);
4309         return;
4310     }
4311
4312     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4313     CHECK_VALIDITY(c->pstream, !name || pa_utf8_valid(name), tag, PA_ERR_INVALID);
4314     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4315     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4316     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4317
4318     if (idx != PA_INVALID_INDEX)
4319         m = pa_idxset_get_by_index(c->protocol->core->modules, idx);
4320     else {
4321         for (m = pa_idxset_first(c->protocol->core->modules, &idx); m; m = pa_idxset_next(c->protocol->core->modules, &idx))
4322             if (strcmp(name, m->name) == 0)
4323                 break;
4324     }
4325
4326     CHECK_VALIDITY(c->pstream, m, tag, PA_ERR_NOEXTENSION);
4327     CHECK_VALIDITY(c->pstream, m->load_once || idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4328
4329     cb = (pa_native_protocol_ext_cb_t) (unsigned long) pa_hashmap_get(c->protocol->extensions, m);
4330     CHECK_VALIDITY(c->pstream, cb, tag, PA_ERR_NOEXTENSION);
4331
4332     if (cb(c->protocol, m, c, tag, t) < 0)
4333         protocol_error(c);
4334 }
4335
4336 static void command_set_card_profile(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4337     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4338     uint32_t idx = PA_INVALID_INDEX;
4339     const char *name = NULL, *profile = NULL;
4340     pa_card *card = NULL;
4341     int ret;
4342
4343     pa_native_connection_assert_ref(c);
4344     pa_assert(t);
4345
4346     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4347         pa_tagstruct_gets(t, &name) < 0 ||
4348         pa_tagstruct_gets(t, &profile) < 0 ||
4349         !pa_tagstruct_eof(t)) {
4350         protocol_error(c);
4351         return;
4352     }
4353
4354     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4355     CHECK_VALIDITY(c->pstream, !name || pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
4356     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4357     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4358     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4359
4360     if (idx != PA_INVALID_INDEX)
4361         card = pa_idxset_get_by_index(c->protocol->core->cards, idx);
4362     else
4363         card = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_CARD);
4364
4365     CHECK_VALIDITY(c->pstream, card, tag, PA_ERR_NOENTITY);
4366
4367     if ((ret = pa_card_set_profile(card, profile, TRUE)) < 0) {
4368         pa_pstream_send_error(c->pstream, tag, -ret);
4369         return;
4370     }
4371
4372     pa_pstream_send_simple_ack(c->pstream, tag);
4373 }
4374
4375 static void command_set_sink_or_source_port(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4376     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4377     uint32_t idx = PA_INVALID_INDEX;
4378     const char *name = NULL, *port = NULL;
4379     int ret;
4380
4381     pa_native_connection_assert_ref(c);
4382     pa_assert(t);
4383
4384     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4385         pa_tagstruct_gets(t, &name) < 0 ||
4386         pa_tagstruct_gets(t, &port) < 0 ||
4387         !pa_tagstruct_eof(t)) {
4388         protocol_error(c);
4389         return;
4390     }
4391
4392     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4393     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);
4394     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4395     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4396     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4397
4398     if (command == PA_COMMAND_SET_SINK_PORT) {
4399         pa_sink *sink;
4400
4401         if (idx != PA_INVALID_INDEX)
4402             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
4403         else
4404             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
4405
4406         CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
4407
4408         if ((ret = pa_sink_set_port(sink, port, TRUE)) < 0) {
4409             pa_pstream_send_error(c->pstream, tag, -ret);
4410             return;
4411         }
4412     } else {
4413         pa_source *source;
4414
4415         pa_assert(command = PA_COMMAND_SET_SOURCE_PORT);
4416
4417         if (idx != PA_INVALID_INDEX)
4418             source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
4419         else
4420             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
4421
4422         CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
4423
4424         if ((ret = pa_source_set_port(source, port, TRUE)) < 0) {
4425             pa_pstream_send_error(c->pstream, tag, -ret);
4426             return;
4427         }
4428     }
4429
4430     pa_pstream_send_simple_ack(c->pstream, tag);
4431 }
4432
4433 /*** pstream callbacks ***/
4434
4435 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
4436     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4437
4438     pa_assert(p);
4439     pa_assert(packet);
4440     pa_native_connection_assert_ref(c);
4441
4442     if (pa_pdispatch_run(c->pdispatch, packet, creds, c) < 0) {
4443         pa_log("invalid packet.");
4444         native_connection_unlink(c);
4445     }
4446 }
4447
4448 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) {
4449     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4450     output_stream *stream;
4451
4452     pa_assert(p);
4453     pa_assert(chunk);
4454     pa_native_connection_assert_ref(c);
4455
4456     if (!(stream = OUTPUT_STREAM(pa_idxset_get_by_index(c->output_streams, channel)))) {
4457         pa_log_debug("Client sent block for invalid stream.");
4458         /* Ignoring */
4459         return;
4460     }
4461
4462 /*     pa_log("got %lu bytes", (unsigned long) chunk->length); */
4463
4464     if (playback_stream_isinstance(stream)) {
4465         playback_stream *ps = PLAYBACK_STREAM(stream);
4466
4467         if (chunk->memblock) {
4468             if (seek != PA_SEEK_RELATIVE || offset != 0)
4469                 pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_SEEK, PA_UINT_TO_PTR(seek), offset, NULL, NULL);
4470
4471             pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
4472         } else
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->length, NULL, NULL);
4474
4475     } else {
4476         upload_stream *u = UPLOAD_STREAM(stream);
4477         size_t l;
4478
4479         if (!u->memchunk.memblock) {
4480             if (u->length == chunk->length && chunk->memblock) {
4481                 u->memchunk = *chunk;
4482                 pa_memblock_ref(u->memchunk.memblock);
4483                 u->length = 0;
4484             } else {
4485                 u->memchunk.memblock = pa_memblock_new(c->protocol->core->mempool, u->length);
4486                 u->memchunk.index = u->memchunk.length = 0;
4487             }
4488         }
4489
4490         pa_assert(u->memchunk.memblock);
4491
4492         l = u->length;
4493         if (l > chunk->length)
4494             l = chunk->length;
4495
4496         if (l > 0) {
4497             void *dst;
4498             dst = pa_memblock_acquire(u->memchunk.memblock);
4499
4500             if (chunk->memblock) {
4501                 void *src;
4502                 src = pa_memblock_acquire(chunk->memblock);
4503
4504                 memcpy((uint8_t*) dst + u->memchunk.index + u->memchunk.length,
4505                        (uint8_t*) src + chunk->index, l);
4506
4507                 pa_memblock_release(chunk->memblock);
4508             } else
4509                 pa_silence_memory((uint8_t*) dst + u->memchunk.index + u->memchunk.length, l, &u->sample_spec);
4510
4511             pa_memblock_release(u->memchunk.memblock);
4512
4513             u->memchunk.length += l;
4514             u->length -= l;
4515         }
4516     }
4517 }
4518
4519 static void pstream_die_callback(pa_pstream *p, void *userdata) {
4520     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4521
4522     pa_assert(p);
4523     pa_native_connection_assert_ref(c);
4524
4525     native_connection_unlink(c);
4526     pa_log_info("Connection died.");
4527 }
4528
4529 static void pstream_drain_callback(pa_pstream *p, void *userdata) {
4530     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4531
4532     pa_assert(p);
4533     pa_native_connection_assert_ref(c);
4534
4535     native_connection_send_memblock(c);
4536 }
4537
4538 static void pstream_revoke_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
4539     pa_thread_mq *q;
4540
4541     if (!(q = pa_thread_mq_get()))
4542         pa_pstream_send_revoke(p, block_id);
4543     else
4544         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_REVOKE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
4545 }
4546
4547 static void pstream_release_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
4548     pa_thread_mq *q;
4549
4550     if (!(q = pa_thread_mq_get()))
4551         pa_pstream_send_release(p, block_id);
4552     else
4553         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_RELEASE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
4554 }
4555
4556 /*** client callbacks ***/
4557
4558 static void client_kill_cb(pa_client *c) {
4559     pa_assert(c);
4560
4561     native_connection_unlink(PA_NATIVE_CONNECTION(c->userdata));
4562     pa_log_info("Connection killed.");
4563 }
4564
4565 static void client_send_event_cb(pa_client *client, const char*event, pa_proplist *pl) {
4566     pa_tagstruct *t;
4567     pa_native_connection *c;
4568
4569     pa_assert(client);
4570     c = PA_NATIVE_CONNECTION(client->userdata);
4571     pa_native_connection_assert_ref(c);
4572
4573     if (c->version < 15)
4574       return;
4575
4576     t = pa_tagstruct_new(NULL, 0);
4577     pa_tagstruct_putu32(t, PA_COMMAND_CLIENT_EVENT);
4578     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
4579     pa_tagstruct_puts(t, event);
4580     pa_tagstruct_put_proplist(t, pl);
4581     pa_pstream_send_tagstruct(c->pstream, t);
4582 }
4583
4584 /*** module entry points ***/
4585
4586 static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *t, void *userdata) {
4587     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4588
4589     pa_assert(m);
4590     pa_native_connection_assert_ref(c);
4591     pa_assert(c->auth_timeout_event == e);
4592
4593     if (!c->authorized) {
4594         native_connection_unlink(c);
4595         pa_log_info("Connection terminated due to authentication timeout.");
4596     }
4597 }
4598
4599 void pa_native_protocol_connect(pa_native_protocol *p, pa_iochannel *io, pa_native_options *o) {
4600     pa_native_connection *c;
4601     char pname[128];
4602     pa_client *client;
4603     pa_client_new_data data;
4604
4605     pa_assert(p);
4606     pa_assert(io);
4607     pa_assert(o);
4608
4609     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
4610         pa_log_warn("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
4611         pa_iochannel_free(io);
4612         return;
4613     }
4614
4615     pa_client_new_data_init(&data);
4616     data.module = o->module;
4617     data.driver = __FILE__;
4618     pa_iochannel_socket_peer_to_string(io, pname, sizeof(pname));
4619     pa_proplist_setf(data.proplist, PA_PROP_APPLICATION_NAME, "Native client (%s)", pname);
4620     pa_proplist_sets(data.proplist, "native-protocol.peer", pname);
4621     client = pa_client_new(p->core, &data);
4622     pa_client_new_data_done(&data);
4623
4624     if (!client)
4625         return;
4626
4627     c = pa_msgobject_new(pa_native_connection);
4628     c->parent.parent.free = native_connection_free;
4629     c->parent.process_msg = native_connection_process_msg;
4630     c->protocol = p;
4631     c->options = pa_native_options_ref(o);
4632     c->authorized = FALSE;
4633
4634     if (o->auth_anonymous) {
4635         pa_log_info("Client authenticated anonymously.");
4636         c->authorized = TRUE;
4637     }
4638
4639     if (!c->authorized &&
4640         o->auth_ip_acl &&
4641         pa_ip_acl_check(o->auth_ip_acl, pa_iochannel_get_recv_fd(io)) > 0) {
4642
4643         pa_log_info("Client authenticated by IP ACL.");
4644         c->authorized = TRUE;
4645     }
4646
4647     if (!c->authorized)
4648         c->auth_timeout_event = pa_core_rttime_new(p->core, pa_rtclock_now() + AUTH_TIMEOUT, auth_timeout, c);
4649     else
4650         c->auth_timeout_event = NULL;
4651
4652     c->is_local = pa_iochannel_socket_is_local(io);
4653     c->version = 8;
4654
4655     c->client = client;
4656     c->client->kill = client_kill_cb;
4657     c->client->send_event = client_send_event_cb;
4658     c->client->userdata = c;
4659
4660     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->mempool);
4661     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
4662     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
4663     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
4664     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
4665     pa_pstream_set_revoke_callback(c->pstream, pstream_revoke_callback, c);
4666     pa_pstream_set_release_callback(c->pstream, pstream_release_callback, c);
4667
4668     c->pdispatch = pa_pdispatch_new(p->core->mainloop, TRUE, command_table, PA_COMMAND_MAX);
4669
4670     c->record_streams = pa_idxset_new(NULL, NULL);
4671     c->output_streams = pa_idxset_new(NULL, NULL);
4672
4673     c->rrobin_index = PA_IDXSET_INVALID;
4674     c->subscription = NULL;
4675
4676     pa_idxset_put(p->connections, c, NULL);
4677
4678 #ifdef HAVE_CREDS
4679     if (pa_iochannel_creds_supported(io))
4680         pa_iochannel_creds_enable(io);
4681 #endif
4682
4683     pa_hook_fire(&p->hooks[PA_NATIVE_HOOK_CONNECTION_PUT], c);
4684 }
4685
4686 void pa_native_protocol_disconnect(pa_native_protocol *p, pa_module *m) {
4687     pa_native_connection *c;
4688     void *state = NULL;
4689
4690     pa_assert(p);
4691     pa_assert(m);
4692
4693     while ((c = pa_idxset_iterate(p->connections, &state, NULL)))
4694         if (c->options->module == m)
4695             native_connection_unlink(c);
4696 }
4697
4698 static pa_native_protocol* native_protocol_new(pa_core *c) {
4699     pa_native_protocol *p;
4700     pa_native_hook_t h;
4701
4702     pa_assert(c);
4703
4704     p = pa_xnew(pa_native_protocol, 1);
4705     PA_REFCNT_INIT(p);
4706     p->core = c;
4707     p->connections = pa_idxset_new(NULL, NULL);
4708
4709     p->servers = NULL;
4710
4711     p->extensions = pa_hashmap_new(pa_idxset_trivial_hash_func, pa_idxset_trivial_compare_func);
4712
4713     for (h = 0; h < PA_NATIVE_HOOK_MAX; h++)
4714         pa_hook_init(&p->hooks[h], p);
4715
4716     pa_assert_se(pa_shared_set(c, "native-protocol", p) >= 0);
4717
4718     return p;
4719 }
4720
4721 pa_native_protocol* pa_native_protocol_get(pa_core *c) {
4722     pa_native_protocol *p;
4723
4724     if ((p = pa_shared_get(c, "native-protocol")))
4725         return pa_native_protocol_ref(p);
4726
4727     return native_protocol_new(c);
4728 }
4729
4730 pa_native_protocol* pa_native_protocol_ref(pa_native_protocol *p) {
4731     pa_assert(p);
4732     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4733
4734     PA_REFCNT_INC(p);
4735
4736     return p;
4737 }
4738
4739 void pa_native_protocol_unref(pa_native_protocol *p) {
4740     pa_native_connection *c;
4741     pa_native_hook_t h;
4742
4743     pa_assert(p);
4744     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4745
4746     if (PA_REFCNT_DEC(p) > 0)
4747         return;
4748
4749     while ((c = pa_idxset_first(p->connections, NULL)))
4750         native_connection_unlink(c);
4751
4752     pa_idxset_free(p->connections, NULL, NULL);
4753
4754     pa_strlist_free(p->servers);
4755
4756     for (h = 0; h < PA_NATIVE_HOOK_MAX; h++)
4757         pa_hook_done(&p->hooks[h]);
4758
4759     pa_hashmap_free(p->extensions, NULL, NULL);
4760
4761     pa_assert_se(pa_shared_remove(p->core, "native-protocol") >= 0);
4762
4763     pa_xfree(p);
4764 }
4765
4766 void pa_native_protocol_add_server_string(pa_native_protocol *p, const char *name) {
4767     pa_assert(p);
4768     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4769     pa_assert(name);
4770
4771     p->servers = pa_strlist_prepend(p->servers, name);
4772
4773     pa_hook_fire(&p->hooks[PA_NATIVE_HOOK_SERVERS_CHANGED], p->servers);
4774 }
4775
4776 void pa_native_protocol_remove_server_string(pa_native_protocol *p, const char *name) {
4777     pa_assert(p);
4778     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4779     pa_assert(name);
4780
4781     p->servers = pa_strlist_remove(p->servers, name);
4782
4783     pa_hook_fire(&p->hooks[PA_NATIVE_HOOK_SERVERS_CHANGED], p->servers);
4784 }
4785
4786 pa_hook *pa_native_protocol_hooks(pa_native_protocol *p) {
4787     pa_assert(p);
4788     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4789
4790     return p->hooks;
4791 }
4792
4793 pa_strlist *pa_native_protocol_servers(pa_native_protocol *p) {
4794     pa_assert(p);
4795     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4796
4797     return p->servers;
4798 }
4799
4800 int pa_native_protocol_install_ext(pa_native_protocol *p, pa_module *m, pa_native_protocol_ext_cb_t cb) {
4801     pa_assert(p);
4802     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4803     pa_assert(m);
4804     pa_assert(cb);
4805     pa_assert(!pa_hashmap_get(p->extensions, m));
4806
4807     pa_assert_se(pa_hashmap_put(p->extensions, m, (void*) (unsigned long) cb) == 0);
4808     return 0;
4809 }
4810
4811 void pa_native_protocol_remove_ext(pa_native_protocol *p, pa_module *m) {
4812     pa_assert(p);
4813     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4814     pa_assert(m);
4815
4816     pa_assert_se(pa_hashmap_remove(p->extensions, m));
4817 }
4818
4819 pa_native_options* pa_native_options_new(void) {
4820     pa_native_options *o;
4821
4822     o = pa_xnew0(pa_native_options, 1);
4823     PA_REFCNT_INIT(o);
4824
4825     return o;
4826 }
4827
4828 pa_native_options* pa_native_options_ref(pa_native_options *o) {
4829     pa_assert(o);
4830     pa_assert(PA_REFCNT_VALUE(o) >= 1);
4831
4832     PA_REFCNT_INC(o);
4833
4834     return o;
4835 }
4836
4837 void pa_native_options_unref(pa_native_options *o) {
4838     pa_assert(o);
4839     pa_assert(PA_REFCNT_VALUE(o) >= 1);
4840
4841     if (PA_REFCNT_DEC(o) > 0)
4842         return;
4843
4844     pa_xfree(o->auth_group);
4845
4846     if (o->auth_ip_acl)
4847         pa_ip_acl_free(o->auth_ip_acl);
4848
4849     if (o->auth_cookie)
4850         pa_auth_cookie_unref(o->auth_cookie);
4851
4852     pa_xfree(o);
4853 }
4854
4855 int pa_native_options_parse(pa_native_options *o, pa_core *c, pa_modargs *ma) {
4856     pa_bool_t enabled;
4857     const char *acl;
4858
4859     pa_assert(o);
4860     pa_assert(PA_REFCNT_VALUE(o) >= 1);
4861     pa_assert(ma);
4862
4863     if (pa_modargs_get_value_boolean(ma, "auth-anonymous", &o->auth_anonymous) < 0) {
4864         pa_log("auth-anonymous= expects a boolean argument.");
4865         return -1;
4866     }
4867
4868     enabled = TRUE;
4869     if (pa_modargs_get_value_boolean(ma, "auth-group-enabled", &enabled) < 0) {
4870         pa_log("auth-group-enabled= expects a boolean argument.");
4871         return -1;
4872     }
4873
4874     pa_xfree(o->auth_group);
4875     o->auth_group = enabled ? pa_xstrdup(pa_modargs_get_value(ma, "auth-group", pa_in_system_mode() ? PA_ACCESS_GROUP : NULL)) : NULL;
4876
4877 #ifndef HAVE_CREDS
4878     if (o->auth_group)
4879         pa_log_warn("Authentication group configured, but not available on local system. Ignoring.");
4880 #endif
4881
4882     if ((acl = pa_modargs_get_value(ma, "auth-ip-acl", NULL))) {
4883         pa_ip_acl *ipa;
4884
4885         if (!(ipa = pa_ip_acl_new(acl))) {
4886             pa_log("Failed to parse IP ACL '%s'", acl);
4887             return -1;
4888         }
4889
4890         if (o->auth_ip_acl)
4891             pa_ip_acl_free(o->auth_ip_acl);
4892
4893         o->auth_ip_acl = ipa;
4894     }
4895
4896     enabled = TRUE;
4897     if (pa_modargs_get_value_boolean(ma, "auth-cookie-enabled", &enabled) < 0) {
4898         pa_log("auth-cookie-enabled= expects a boolean argument.");
4899         return -1;
4900     }
4901
4902     if (o->auth_cookie)
4903         pa_auth_cookie_unref(o->auth_cookie);
4904
4905     if (enabled) {
4906         const char *cn;
4907
4908         /* The new name for this is 'auth-cookie', for compat reasons
4909          * we check the old name too */
4910         if (!(cn = pa_modargs_get_value(ma, "auth-cookie", NULL)))
4911             if (!(cn = pa_modargs_get_value(ma, "cookie", NULL)))
4912                 cn = PA_NATIVE_COOKIE_FILE;
4913
4914         if (!(o->auth_cookie = pa_auth_cookie_get(c, cn, PA_NATIVE_COOKIE_LENGTH)))
4915             return -1;
4916
4917     } else
4918           o->auth_cookie = NULL;
4919
4920     return 0;
4921 }
4922
4923 pa_pstream* pa_native_connection_get_pstream(pa_native_connection *c) {
4924     pa_native_connection_assert_ref(c);
4925
4926     return c->pstream;
4927 }
4928
4929 pa_client* pa_native_connection_get_client(pa_native_connection *c) {
4930    pa_native_connection_assert_ref(c);
4931
4932    return c->client;
4933 }