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