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