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