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 & ~PA_SINK_SHARE_VOLUME_WITH_MASTER,
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     pa_bool_t has_volume = FALSE;
3060
3061     pa_assert(t);
3062     pa_sink_input_assert_ref(s);
3063
3064     fixup_sample_spec(c, &fixed_ss, &s->sample_spec);
3065
3066     has_volume = pa_sink_input_is_volume_readable(s);
3067     if (has_volume)
3068         pa_sink_input_get_volume(s, &v, TRUE);
3069     else
3070         pa_cvolume_reset(&v, fixed_ss.channels);
3071
3072     pa_tagstruct_putu32(t, s->index);
3073     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME)));
3074     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
3075     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
3076     pa_tagstruct_putu32(t, s->sink->index);
3077     pa_tagstruct_put_sample_spec(t, &fixed_ss);
3078     pa_tagstruct_put_channel_map(t, &s->channel_map);
3079     pa_tagstruct_put_cvolume(t, &v);
3080     pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s, &sink_latency));
3081     pa_tagstruct_put_usec(t, sink_latency);
3082     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
3083     pa_tagstruct_puts(t, s->driver);
3084     if (c->version >= 11)
3085         pa_tagstruct_put_boolean(t, pa_sink_input_get_mute(s));
3086     if (c->version >= 13)
3087         pa_tagstruct_put_proplist(t, s->proplist);
3088     if (c->version >= 19)
3089         pa_tagstruct_put_boolean(t, (pa_sink_input_get_state(s) == PA_SINK_INPUT_CORKED));
3090     if (c->version >= 20) {
3091         pa_tagstruct_put_boolean(t, has_volume);
3092         pa_tagstruct_put_boolean(t, has_volume ? !pa_sink_input_is_volume_writable(s) : FALSE);
3093     }
3094 }
3095
3096 static void source_output_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_source_output *s) {
3097     pa_sample_spec fixed_ss;
3098     pa_usec_t source_latency;
3099
3100     pa_assert(t);
3101     pa_source_output_assert_ref(s);
3102
3103     fixup_sample_spec(c, &fixed_ss, &s->sample_spec);
3104
3105     pa_tagstruct_putu32(t, s->index);
3106     pa_tagstruct_puts(t, pa_strnull(pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME)));
3107     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
3108     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
3109     pa_tagstruct_putu32(t, s->source->index);
3110     pa_tagstruct_put_sample_spec(t, &fixed_ss);
3111     pa_tagstruct_put_channel_map(t, &s->channel_map);
3112     pa_tagstruct_put_usec(t, pa_source_output_get_latency(s, &source_latency));
3113     pa_tagstruct_put_usec(t, source_latency);
3114     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_source_output_get_resample_method(s)));
3115     pa_tagstruct_puts(t, s->driver);
3116     if (c->version >= 13)
3117         pa_tagstruct_put_proplist(t, s->proplist);
3118     if (c->version >= 19)
3119         pa_tagstruct_put_boolean(t, (pa_source_output_get_state(s) == PA_SOURCE_OUTPUT_CORKED));
3120 }
3121
3122 static void scache_fill_tagstruct(pa_native_connection *c, pa_tagstruct *t, pa_scache_entry *e) {
3123     pa_sample_spec fixed_ss;
3124     pa_cvolume v;
3125
3126     pa_assert(t);
3127     pa_assert(e);
3128
3129     if (e->memchunk.memblock)
3130         fixup_sample_spec(c, &fixed_ss, &e->sample_spec);
3131     else
3132         memset(&fixed_ss, 0, sizeof(fixed_ss));
3133
3134     pa_tagstruct_putu32(t, e->index);
3135     pa_tagstruct_puts(t, e->name);
3136
3137     if (e->volume_is_set)
3138         v = e->volume;
3139     else
3140         pa_cvolume_init(&v);
3141
3142     pa_tagstruct_put_cvolume(t, &v);
3143     pa_tagstruct_put_usec(t, e->memchunk.memblock ? pa_bytes_to_usec(e->memchunk.length, &e->sample_spec) : 0);
3144     pa_tagstruct_put_sample_spec(t, &fixed_ss);
3145     pa_tagstruct_put_channel_map(t, &e->channel_map);
3146     pa_tagstruct_putu32(t, (uint32_t) e->memchunk.length);
3147     pa_tagstruct_put_boolean(t, e->lazy);
3148     pa_tagstruct_puts(t, e->filename);
3149
3150     if (c->version >= 13)
3151         pa_tagstruct_put_proplist(t, e->proplist);
3152 }
3153
3154 static void command_get_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3155     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3156     uint32_t idx;
3157     pa_sink *sink = NULL;
3158     pa_source *source = NULL;
3159     pa_client *client = NULL;
3160     pa_card *card = NULL;
3161     pa_module *module = NULL;
3162     pa_sink_input *si = NULL;
3163     pa_source_output *so = NULL;
3164     pa_scache_entry *sce = NULL;
3165     const char *name = NULL;
3166     pa_tagstruct *reply;
3167
3168     pa_native_connection_assert_ref(c);
3169     pa_assert(t);
3170
3171     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3172         (command != PA_COMMAND_GET_CLIENT_INFO &&
3173          command != PA_COMMAND_GET_MODULE_INFO &&
3174          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
3175          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
3176          pa_tagstruct_gets(t, &name) < 0) ||
3177         !pa_tagstruct_eof(t)) {
3178         protocol_error(c);
3179         return;
3180     }
3181
3182     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3183     CHECK_VALIDITY(c->pstream, !name ||
3184                    (command == PA_COMMAND_GET_SINK_INFO &&
3185                     pa_namereg_is_valid_name_or_wildcard(name, PA_NAMEREG_SINK)) ||
3186                    (command == PA_COMMAND_GET_SOURCE_INFO &&
3187                     pa_namereg_is_valid_name_or_wildcard(name, PA_NAMEREG_SOURCE)) ||
3188                    pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
3189     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
3190     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
3191     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3192
3193     if (command == PA_COMMAND_GET_SINK_INFO) {
3194         if (idx != PA_INVALID_INDEX)
3195             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3196         else
3197             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
3198     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
3199         if (idx != PA_INVALID_INDEX)
3200             source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3201         else
3202             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
3203     } else if (command == PA_COMMAND_GET_CARD_INFO) {
3204         if (idx != PA_INVALID_INDEX)
3205             card = pa_idxset_get_by_index(c->protocol->core->cards, idx);
3206         else
3207             card = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_CARD);
3208     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
3209         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
3210     else if (command == PA_COMMAND_GET_MODULE_INFO)
3211         module = pa_idxset_get_by_index(c->protocol->core->modules, idx);
3212     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
3213         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3214     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
3215         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
3216     else {
3217         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO);
3218         if (idx != PA_INVALID_INDEX)
3219             sce = pa_idxset_get_by_index(c->protocol->core->scache, idx);
3220         else
3221             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE);
3222     }
3223
3224     if (!sink && !source && !client && !card && !module && !si && !so && !sce) {
3225         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
3226         return;
3227     }
3228
3229     reply = reply_new(tag);
3230     if (sink)
3231         sink_fill_tagstruct(c, reply, sink);
3232     else if (source)
3233         source_fill_tagstruct(c, reply, source);
3234     else if (client)
3235         client_fill_tagstruct(c, reply, client);
3236     else if (card)
3237         card_fill_tagstruct(c, reply, card);
3238     else if (module)
3239         module_fill_tagstruct(c, reply, module);
3240     else if (si)
3241         sink_input_fill_tagstruct(c, reply, si);
3242     else if (so)
3243         source_output_fill_tagstruct(c, reply, so);
3244     else
3245         scache_fill_tagstruct(c, reply, sce);
3246     pa_pstream_send_tagstruct(c->pstream, reply);
3247 }
3248
3249 static void command_get_info_list(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3250     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3251     pa_idxset *i;
3252     uint32_t idx;
3253     void *p;
3254     pa_tagstruct *reply;
3255
3256     pa_native_connection_assert_ref(c);
3257     pa_assert(t);
3258
3259     if (!pa_tagstruct_eof(t)) {
3260         protocol_error(c);
3261         return;
3262     }
3263
3264     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3265
3266     reply = reply_new(tag);
3267
3268     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
3269         i = c->protocol->core->sinks;
3270     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
3271         i = c->protocol->core->sources;
3272     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
3273         i = c->protocol->core->clients;
3274     else if (command == PA_COMMAND_GET_CARD_INFO_LIST)
3275         i = c->protocol->core->cards;
3276     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
3277         i = c->protocol->core->modules;
3278     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
3279         i = c->protocol->core->sink_inputs;
3280     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
3281         i = c->protocol->core->source_outputs;
3282     else {
3283         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
3284         i = c->protocol->core->scache;
3285     }
3286
3287     if (i) {
3288         for (p = pa_idxset_first(i, &idx); p; p = pa_idxset_next(i, &idx)) {
3289             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
3290                 sink_fill_tagstruct(c, reply, p);
3291             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
3292                 source_fill_tagstruct(c, reply, p);
3293             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
3294                 client_fill_tagstruct(c, reply, p);
3295             else if (command == PA_COMMAND_GET_CARD_INFO_LIST)
3296                 card_fill_tagstruct(c, reply, p);
3297             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
3298                 module_fill_tagstruct(c, reply, p);
3299             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
3300                 sink_input_fill_tagstruct(c, reply, p);
3301             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
3302                 source_output_fill_tagstruct(c, reply, p);
3303             else {
3304                 pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
3305                 scache_fill_tagstruct(c, reply, p);
3306             }
3307         }
3308     }
3309
3310     pa_pstream_send_tagstruct(c->pstream, reply);
3311 }
3312
3313 static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3314     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3315     pa_tagstruct *reply;
3316     pa_sink *def_sink;
3317     pa_source *def_source;
3318     pa_sample_spec fixed_ss;
3319     char *h, *u;
3320
3321     pa_native_connection_assert_ref(c);
3322     pa_assert(t);
3323
3324     if (!pa_tagstruct_eof(t)) {
3325         protocol_error(c);
3326         return;
3327     }
3328
3329     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3330
3331     reply = reply_new(tag);
3332     pa_tagstruct_puts(reply, PACKAGE_NAME);
3333     pa_tagstruct_puts(reply, PACKAGE_VERSION);
3334
3335     u = pa_get_user_name_malloc();
3336     pa_tagstruct_puts(reply, u);
3337     pa_xfree(u);
3338
3339     h = pa_get_host_name_malloc();
3340     pa_tagstruct_puts(reply, h);
3341     pa_xfree(h);
3342
3343     fixup_sample_spec(c, &fixed_ss, &c->protocol->core->default_sample_spec);
3344     pa_tagstruct_put_sample_spec(reply, &fixed_ss);
3345
3346     def_sink = pa_namereg_get_default_sink(c->protocol->core);
3347     pa_tagstruct_puts(reply, def_sink ? def_sink->name : NULL);
3348     def_source = pa_namereg_get_default_source(c->protocol->core);
3349     pa_tagstruct_puts(reply, def_source ? def_source->name : NULL);
3350
3351     pa_tagstruct_putu32(reply, c->protocol->core->cookie);
3352
3353     if (c->version >= 15)
3354         pa_tagstruct_put_channel_map(reply, &c->protocol->core->default_channel_map);
3355
3356     pa_pstream_send_tagstruct(c->pstream, reply);
3357 }
3358
3359 static void subscription_cb(pa_core *core, pa_subscription_event_type_t e, uint32_t idx, void *userdata) {
3360     pa_tagstruct *t;
3361     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3362
3363     pa_native_connection_assert_ref(c);
3364
3365     t = pa_tagstruct_new(NULL, 0);
3366     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
3367     pa_tagstruct_putu32(t, (uint32_t) -1);
3368     pa_tagstruct_putu32(t, e);
3369     pa_tagstruct_putu32(t, idx);
3370     pa_pstream_send_tagstruct(c->pstream, t);
3371 }
3372
3373 static void command_subscribe(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3374     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3375     pa_subscription_mask_t m;
3376
3377     pa_native_connection_assert_ref(c);
3378     pa_assert(t);
3379
3380     if (pa_tagstruct_getu32(t, &m) < 0 ||
3381         !pa_tagstruct_eof(t)) {
3382         protocol_error(c);
3383         return;
3384     }
3385
3386     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3387     CHECK_VALIDITY(c->pstream, (m & ~PA_SUBSCRIPTION_MASK_ALL) == 0, tag, PA_ERR_INVALID);
3388
3389     if (c->subscription)
3390         pa_subscription_free(c->subscription);
3391
3392     if (m != 0) {
3393         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
3394         pa_assert(c->subscription);
3395     } else
3396         c->subscription = NULL;
3397
3398     pa_pstream_send_simple_ack(c->pstream, tag);
3399 }
3400
3401 static void command_set_volume(
3402         pa_pdispatch *pd,
3403         uint32_t command,
3404         uint32_t tag,
3405         pa_tagstruct *t,
3406         void *userdata) {
3407
3408     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3409     uint32_t idx;
3410     pa_cvolume volume;
3411     pa_sink *sink = NULL;
3412     pa_source *source = NULL;
3413     pa_sink_input *si = NULL;
3414     const char *name = NULL;
3415     const char *client_name;
3416
3417     pa_native_connection_assert_ref(c);
3418     pa_assert(t);
3419
3420     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3421         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
3422         (command == PA_COMMAND_SET_SOURCE_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
3423         pa_tagstruct_get_cvolume(t, &volume) ||
3424         !pa_tagstruct_eof(t)) {
3425         protocol_error(c);
3426         return;
3427     }
3428
3429     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3430     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);
3431     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
3432     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
3433     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3434     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
3435
3436     switch (command) {
3437
3438         case PA_COMMAND_SET_SINK_VOLUME:
3439             if (idx != PA_INVALID_INDEX)
3440                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3441             else
3442                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
3443             break;
3444
3445         case PA_COMMAND_SET_SOURCE_VOLUME:
3446             if (idx != PA_INVALID_INDEX)
3447                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3448             else
3449                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
3450             break;
3451
3452         case PA_COMMAND_SET_SINK_INPUT_VOLUME:
3453             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3454             break;
3455
3456         default:
3457             pa_assert_not_reached();
3458     }
3459
3460     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
3461
3462     client_name = pa_strnull(pa_proplist_gets(c->client->proplist, PA_PROP_APPLICATION_PROCESS_BINARY));
3463
3464     if (sink) {
3465         CHECK_VALIDITY(c->pstream, volume.channels == 1 || pa_cvolume_compatible(&volume, &sink->sample_spec), tag, PA_ERR_INVALID);
3466
3467         pa_log_debug("Client %s changes volume of sink %s.", client_name, sink->name);
3468         pa_sink_set_volume(sink, &volume, TRUE, TRUE);
3469     } else if (source) {
3470         CHECK_VALIDITY(c->pstream, volume.channels == 1 || pa_cvolume_compatible(&volume, &source->sample_spec), tag, PA_ERR_INVALID);
3471
3472         pa_log_debug("Client %s changes volume of source %s.", client_name, source->name);
3473         pa_source_set_volume(source, &volume, TRUE);
3474     } else if (si) {
3475         CHECK_VALIDITY(c->pstream, pa_sink_input_is_volume_writable(si), tag, PA_ERR_INVALID);
3476         CHECK_VALIDITY(c->pstream, volume.channels == 1 || pa_cvolume_compatible(&volume, &si->sample_spec), tag, PA_ERR_INVALID);
3477
3478         pa_log_debug("Client %s changes volume of sink input %s.",
3479                      client_name,
3480                      pa_strnull(pa_proplist_gets(si->proplist, PA_PROP_MEDIA_NAME)));
3481         pa_sink_input_set_volume(si, &volume, TRUE, TRUE);
3482     }
3483
3484     pa_pstream_send_simple_ack(c->pstream, tag);
3485 }
3486
3487 static void command_set_mute(
3488         pa_pdispatch *pd,
3489         uint32_t command,
3490         uint32_t tag,
3491         pa_tagstruct *t,
3492         void *userdata) {
3493
3494     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3495     uint32_t idx;
3496     pa_bool_t mute;
3497     pa_sink *sink = NULL;
3498     pa_source *source = NULL;
3499     pa_sink_input *si = NULL;
3500     const char *name = NULL, *client_name;
3501
3502     pa_native_connection_assert_ref(c);
3503     pa_assert(t);
3504
3505     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3506         (command == PA_COMMAND_SET_SINK_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
3507         (command == PA_COMMAND_SET_SOURCE_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
3508         pa_tagstruct_get_boolean(t, &mute) ||
3509         !pa_tagstruct_eof(t)) {
3510         protocol_error(c);
3511         return;
3512     }
3513
3514     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3515     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);
3516     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
3517     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
3518     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3519
3520     switch (command) {
3521
3522         case PA_COMMAND_SET_SINK_MUTE:
3523             if (idx != PA_INVALID_INDEX)
3524                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3525             else
3526                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
3527
3528             break;
3529
3530         case PA_COMMAND_SET_SOURCE_MUTE:
3531             if (idx != PA_INVALID_INDEX)
3532                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3533             else
3534                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
3535
3536             break;
3537
3538         case PA_COMMAND_SET_SINK_INPUT_MUTE:
3539             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
3540             break;
3541
3542         default:
3543             pa_assert_not_reached();
3544     }
3545
3546     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
3547
3548     client_name = pa_strnull(pa_proplist_gets(c->client->proplist, PA_PROP_APPLICATION_PROCESS_BINARY));
3549
3550     if (sink) {
3551         pa_log_debug("Client %s changes mute of sink %s.", client_name, sink->name);
3552         pa_sink_set_mute(sink, mute, TRUE);
3553     } else if (source) {
3554         pa_log_debug("Client %s changes mute of source %s.", client_name, source->name);
3555         pa_source_set_mute(source, mute, TRUE);
3556     } else if (si) {
3557         pa_log_debug("Client %s changes mute of sink input %s.",
3558                      client_name,
3559                      pa_strnull(pa_proplist_gets(si->proplist, PA_PROP_MEDIA_NAME)));
3560         pa_sink_input_set_mute(si, mute, TRUE);
3561     }
3562
3563     pa_pstream_send_simple_ack(c->pstream, tag);
3564 }
3565
3566 static void command_cork_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3567     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3568     uint32_t idx;
3569     pa_bool_t b;
3570     playback_stream *s;
3571
3572     pa_native_connection_assert_ref(c);
3573     pa_assert(t);
3574
3575     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3576         pa_tagstruct_get_boolean(t, &b) < 0 ||
3577         !pa_tagstruct_eof(t)) {
3578         protocol_error(c);
3579         return;
3580     }
3581
3582     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3583     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3584     s = pa_idxset_get_by_index(c->output_streams, idx);
3585     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3586     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3587
3588     pa_sink_input_cork(s->sink_input, b);
3589
3590     if (b)
3591         s->is_underrun = TRUE;
3592
3593     pa_pstream_send_simple_ack(c->pstream, tag);
3594 }
3595
3596 static void command_trigger_or_flush_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3597     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3598     uint32_t idx;
3599     playback_stream *s;
3600
3601     pa_native_connection_assert_ref(c);
3602     pa_assert(t);
3603
3604     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3605         !pa_tagstruct_eof(t)) {
3606         protocol_error(c);
3607         return;
3608     }
3609
3610     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3611     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
3612     s = pa_idxset_get_by_index(c->output_streams, idx);
3613     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3614     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3615
3616     switch (command) {
3617         case PA_COMMAND_FLUSH_PLAYBACK_STREAM:
3618             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_FLUSH, NULL, 0, NULL);
3619             break;
3620
3621         case PA_COMMAND_PREBUF_PLAYBACK_STREAM:
3622             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_PREBUF_FORCE, NULL, 0, NULL);
3623             break;
3624
3625         case PA_COMMAND_TRIGGER_PLAYBACK_STREAM:
3626             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_TRIGGER, NULL, 0, NULL);
3627             break;
3628
3629         default:
3630             pa_assert_not_reached();
3631     }
3632
3633     pa_pstream_send_simple_ack(c->pstream, tag);
3634 }
3635
3636 static void command_cork_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3637     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3638     uint32_t idx;
3639     record_stream *s;
3640     pa_bool_t b;
3641
3642     pa_native_connection_assert_ref(c);
3643     pa_assert(t);
3644
3645     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3646         pa_tagstruct_get_boolean(t, &b) < 0 ||
3647         !pa_tagstruct_eof(t)) {
3648         protocol_error(c);
3649         return;
3650     }
3651
3652     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3653     s = pa_idxset_get_by_index(c->record_streams, idx);
3654     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3655
3656     pa_source_output_cork(s->source_output, b);
3657     pa_memblockq_prebuf_force(s->memblockq);
3658     pa_pstream_send_simple_ack(c->pstream, tag);
3659 }
3660
3661 static void command_flush_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3662     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3663     uint32_t idx;
3664     record_stream *s;
3665
3666     pa_native_connection_assert_ref(c);
3667     pa_assert(t);
3668
3669     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3670         !pa_tagstruct_eof(t)) {
3671         protocol_error(c);
3672         return;
3673     }
3674
3675     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3676     s = pa_idxset_get_by_index(c->record_streams, idx);
3677     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3678
3679     pa_memblockq_flush_read(s->memblockq);
3680     pa_pstream_send_simple_ack(c->pstream, tag);
3681 }
3682
3683 static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3684     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3685     uint32_t idx;
3686     pa_buffer_attr a;
3687     pa_tagstruct *reply;
3688
3689     pa_native_connection_assert_ref(c);
3690     pa_assert(t);
3691
3692     memset(&a, 0, sizeof(a));
3693
3694     if (pa_tagstruct_getu32(t, &idx) < 0) {
3695         protocol_error(c);
3696         return;
3697     }
3698
3699     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3700
3701     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR) {
3702         playback_stream *s;
3703         pa_bool_t adjust_latency = FALSE, early_requests = FALSE;
3704
3705         s = pa_idxset_get_by_index(c->output_streams, idx);
3706         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3707         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3708
3709         if (pa_tagstruct_get(
3710                     t,
3711                     PA_TAG_U32, &a.maxlength,
3712                     PA_TAG_U32, &a.tlength,
3713                     PA_TAG_U32, &a.prebuf,
3714                     PA_TAG_U32, &a.minreq,
3715                     PA_TAG_INVALID) < 0 ||
3716             (c->version >= 13 && pa_tagstruct_get_boolean(t, &adjust_latency) < 0) ||
3717             (c->version >= 14 && pa_tagstruct_get_boolean(t, &early_requests) < 0) ||
3718             !pa_tagstruct_eof(t)) {
3719             protocol_error(c);
3720             return;
3721         }
3722
3723         s->adjust_latency = adjust_latency;
3724         s->early_requests = early_requests;
3725         s->buffer_attr = a;
3726
3727         fix_playback_buffer_attr(s);
3728         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);
3729
3730         reply = reply_new(tag);
3731         pa_tagstruct_putu32(reply, s->buffer_attr.maxlength);
3732         pa_tagstruct_putu32(reply, s->buffer_attr.tlength);
3733         pa_tagstruct_putu32(reply, s->buffer_attr.prebuf);
3734         pa_tagstruct_putu32(reply, s->buffer_attr.minreq);
3735
3736         if (c->version >= 13)
3737             pa_tagstruct_put_usec(reply, s->configured_sink_latency);
3738
3739     } else {
3740         record_stream *s;
3741         pa_bool_t adjust_latency = FALSE, early_requests = FALSE;
3742         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR);
3743
3744         s = pa_idxset_get_by_index(c->record_streams, idx);
3745         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3746
3747         if (pa_tagstruct_get(
3748                     t,
3749                     PA_TAG_U32, &a.maxlength,
3750                     PA_TAG_U32, &a.fragsize,
3751                     PA_TAG_INVALID) < 0 ||
3752             (c->version >= 13 && pa_tagstruct_get_boolean(t, &adjust_latency) < 0) ||
3753             (c->version >= 14 && pa_tagstruct_get_boolean(t, &early_requests) < 0) ||
3754             !pa_tagstruct_eof(t)) {
3755             protocol_error(c);
3756             return;
3757         }
3758
3759         s->adjust_latency = adjust_latency;
3760         s->early_requests = early_requests;
3761         s->buffer_attr = a;
3762
3763         fix_record_buffer_attr_pre(s);
3764         pa_memblockq_set_maxlength(s->memblockq, s->buffer_attr.maxlength);
3765         pa_memblockq_get_attr(s->memblockq, &s->buffer_attr);
3766         fix_record_buffer_attr_post(s);
3767
3768         reply = reply_new(tag);
3769         pa_tagstruct_putu32(reply, s->buffer_attr.maxlength);
3770         pa_tagstruct_putu32(reply, s->buffer_attr.fragsize);
3771
3772         if (c->version >= 13)
3773             pa_tagstruct_put_usec(reply, s->configured_source_latency);
3774     }
3775
3776     pa_pstream_send_tagstruct(c->pstream, reply);
3777 }
3778
3779 static void command_update_stream_sample_rate(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3780     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3781     uint32_t idx;
3782     uint32_t rate;
3783
3784     pa_native_connection_assert_ref(c);
3785     pa_assert(t);
3786
3787     if (pa_tagstruct_getu32(t, &idx) < 0 ||
3788         pa_tagstruct_getu32(t, &rate) < 0 ||
3789         !pa_tagstruct_eof(t)) {
3790         protocol_error(c);
3791         return;
3792     }
3793
3794     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3795     CHECK_VALIDITY(c->pstream, rate > 0 && rate <= PA_RATE_MAX, tag, PA_ERR_INVALID);
3796
3797     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE) {
3798         playback_stream *s;
3799
3800         s = pa_idxset_get_by_index(c->output_streams, idx);
3801         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3802         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3803
3804         pa_sink_input_set_rate(s->sink_input, rate);
3805
3806     } else {
3807         record_stream *s;
3808         pa_assert(command == PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE);
3809
3810         s = pa_idxset_get_by_index(c->record_streams, idx);
3811         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3812
3813         pa_source_output_set_rate(s->source_output, rate);
3814     }
3815
3816     pa_pstream_send_simple_ack(c->pstream, tag);
3817 }
3818
3819 static void command_update_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3820     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3821     uint32_t idx;
3822     uint32_t mode;
3823     pa_proplist *p;
3824
3825     pa_native_connection_assert_ref(c);
3826     pa_assert(t);
3827
3828     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3829
3830     p = pa_proplist_new();
3831
3832     if (command == PA_COMMAND_UPDATE_CLIENT_PROPLIST) {
3833
3834         if (pa_tagstruct_getu32(t, &mode) < 0 ||
3835             pa_tagstruct_get_proplist(t, p) < 0 ||
3836             !pa_tagstruct_eof(t)) {
3837             protocol_error(c);
3838             pa_proplist_free(p);
3839             return;
3840         }
3841
3842     } else {
3843
3844         if (pa_tagstruct_getu32(t, &idx) < 0 ||
3845             pa_tagstruct_getu32(t, &mode) < 0 ||
3846             pa_tagstruct_get_proplist(t, p) < 0 ||
3847             !pa_tagstruct_eof(t)) {
3848             protocol_error(c);
3849             pa_proplist_free(p);
3850             return;
3851         }
3852     }
3853
3854     if (!(mode == PA_UPDATE_SET || mode == PA_UPDATE_MERGE || mode == PA_UPDATE_REPLACE)) {
3855         pa_proplist_free(p);
3856         CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_INVALID);
3857     }
3858
3859     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST) {
3860         playback_stream *s;
3861
3862         s = pa_idxset_get_by_index(c->output_streams, idx);
3863         if (!s || !playback_stream_isinstance(s)) {
3864             pa_proplist_free(p);
3865             CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_NOENTITY);
3866         }
3867         pa_sink_input_update_proplist(s->sink_input, mode, p);
3868
3869     } else if (command == PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST) {
3870         record_stream *s;
3871
3872         if (!(s = pa_idxset_get_by_index(c->record_streams, idx))) {
3873             pa_proplist_free(p);
3874             CHECK_VALIDITY(c->pstream, FALSE, tag, PA_ERR_NOENTITY);
3875         }
3876         pa_source_output_update_proplist(s->source_output, mode, p);
3877
3878     } else {
3879         pa_assert(command == PA_COMMAND_UPDATE_CLIENT_PROPLIST);
3880
3881         pa_client_update_proplist(c->client, mode, p);
3882     }
3883
3884     pa_pstream_send_simple_ack(c->pstream, tag);
3885     pa_proplist_free(p);
3886 }
3887
3888 static void command_remove_proplist(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3889     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3890     uint32_t idx;
3891     unsigned changed = 0;
3892     pa_proplist *p;
3893     pa_strlist *l = NULL;
3894
3895     pa_native_connection_assert_ref(c);
3896     pa_assert(t);
3897
3898     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3899
3900     if (command != PA_COMMAND_REMOVE_CLIENT_PROPLIST) {
3901
3902         if (pa_tagstruct_getu32(t, &idx) < 0) {
3903             protocol_error(c);
3904             return;
3905         }
3906     }
3907
3908     if (command == PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST) {
3909         playback_stream *s;
3910
3911         s = pa_idxset_get_by_index(c->output_streams, idx);
3912         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3913         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
3914
3915         p = s->sink_input->proplist;
3916
3917     } else if (command == PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST) {
3918         record_stream *s;
3919
3920         s = pa_idxset_get_by_index(c->record_streams, idx);
3921         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
3922
3923         p = s->source_output->proplist;
3924     } else {
3925         pa_assert(command == PA_COMMAND_REMOVE_CLIENT_PROPLIST);
3926
3927         p = c->client->proplist;
3928     }
3929
3930     for (;;) {
3931         const char *k;
3932
3933         if (pa_tagstruct_gets(t, &k) < 0) {
3934             protocol_error(c);
3935             pa_strlist_free(l);
3936             return;
3937         }
3938
3939         if (!k)
3940             break;
3941
3942         l = pa_strlist_prepend(l, k);
3943     }
3944
3945     if (!pa_tagstruct_eof(t)) {
3946         protocol_error(c);
3947         pa_strlist_free(l);
3948         return;
3949     }
3950
3951     for (;;) {
3952         char *z;
3953
3954         l = pa_strlist_pop(l, &z);
3955
3956         if (!z)
3957             break;
3958
3959         changed += (unsigned) (pa_proplist_unset(p, z) >= 0);
3960         pa_xfree(z);
3961     }
3962
3963     pa_pstream_send_simple_ack(c->pstream, tag);
3964
3965     if (changed) {
3966         if (command == PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST) {
3967             playback_stream *s;
3968
3969             s = pa_idxset_get_by_index(c->output_streams, idx);
3970             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->sink_input->index);
3971
3972         } else if (command == PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST) {
3973             record_stream *s;
3974
3975             s = pa_idxset_get_by_index(c->record_streams, idx);
3976             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_CHANGE, s->source_output->index);
3977
3978         } else {
3979             pa_assert(command == PA_COMMAND_REMOVE_CLIENT_PROPLIST);
3980             pa_subscription_post(c->protocol->core, PA_SUBSCRIPTION_EVENT_CLIENT|PA_SUBSCRIPTION_EVENT_CHANGE, c->client->index);
3981         }
3982     }
3983 }
3984
3985 static void command_set_default_sink_or_source(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
3986     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
3987     const char *s;
3988
3989     pa_native_connection_assert_ref(c);
3990     pa_assert(t);
3991
3992     if (pa_tagstruct_gets(t, &s) < 0 ||
3993         !pa_tagstruct_eof(t)) {
3994         protocol_error(c);
3995         return;
3996     }
3997
3998     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
3999     CHECK_VALIDITY(c->pstream, !s || pa_namereg_is_valid_name(s), tag, PA_ERR_INVALID);
4000
4001     if (command == PA_COMMAND_SET_DEFAULT_SOURCE) {
4002         pa_source *source;
4003
4004         source = pa_namereg_get(c->protocol->core, s, PA_NAMEREG_SOURCE);
4005         CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
4006
4007         pa_namereg_set_default_source(c->protocol->core, source);
4008     } else {
4009         pa_sink *sink;
4010         pa_assert(command == PA_COMMAND_SET_DEFAULT_SINK);
4011
4012         sink = pa_namereg_get(c->protocol->core, s, PA_NAMEREG_SINK);
4013         CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
4014
4015         pa_namereg_set_default_sink(c->protocol->core, sink);
4016     }
4017
4018     pa_pstream_send_simple_ack(c->pstream, tag);
4019 }
4020
4021 static void command_set_stream_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4022     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4023     uint32_t idx;
4024     const char *name;
4025
4026     pa_native_connection_assert_ref(c);
4027     pa_assert(t);
4028
4029     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4030         pa_tagstruct_gets(t, &name) < 0 ||
4031         !pa_tagstruct_eof(t)) {
4032         protocol_error(c);
4033         return;
4034     }
4035
4036     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4037     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
4038
4039     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
4040         playback_stream *s;
4041
4042         s = pa_idxset_get_by_index(c->output_streams, idx);
4043         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4044         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
4045
4046         pa_sink_input_set_name(s->sink_input, name);
4047
4048     } else {
4049         record_stream *s;
4050         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_NAME);
4051
4052         s = pa_idxset_get_by_index(c->record_streams, idx);
4053         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4054
4055         pa_source_output_set_name(s->source_output, name);
4056     }
4057
4058     pa_pstream_send_simple_ack(c->pstream, tag);
4059 }
4060
4061 static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4062     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4063     uint32_t idx;
4064
4065     pa_native_connection_assert_ref(c);
4066     pa_assert(t);
4067
4068     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4069         !pa_tagstruct_eof(t)) {
4070         protocol_error(c);
4071         return;
4072     }
4073
4074     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4075
4076     if (command == PA_COMMAND_KILL_CLIENT) {
4077         pa_client *client;
4078
4079         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
4080         CHECK_VALIDITY(c->pstream, client, tag, PA_ERR_NOENTITY);
4081
4082         pa_native_connection_ref(c);
4083         pa_client_kill(client);
4084
4085     } else if (command == PA_COMMAND_KILL_SINK_INPUT) {
4086         pa_sink_input *s;
4087
4088         s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
4089         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4090
4091         pa_native_connection_ref(c);
4092         pa_sink_input_kill(s);
4093     } else {
4094         pa_source_output *s;
4095
4096         pa_assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
4097
4098         s = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
4099         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
4100
4101         pa_native_connection_ref(c);
4102         pa_source_output_kill(s);
4103     }
4104
4105     pa_pstream_send_simple_ack(c->pstream, tag);
4106     pa_native_connection_unref(c);
4107 }
4108
4109 static void command_load_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4110     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4111     pa_module *m;
4112     const char *name, *argument;
4113     pa_tagstruct *reply;
4114
4115     pa_native_connection_assert_ref(c);
4116     pa_assert(t);
4117
4118     if (pa_tagstruct_gets(t, &name) < 0 ||
4119         pa_tagstruct_gets(t, &argument) < 0 ||
4120         !pa_tagstruct_eof(t)) {
4121         protocol_error(c);
4122         return;
4123     }
4124
4125     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4126     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name) && !strchr(name, '/'), tag, PA_ERR_INVALID);
4127     CHECK_VALIDITY(c->pstream, !argument || pa_utf8_valid(argument), tag, PA_ERR_INVALID);
4128
4129     if (!(m = pa_module_load(c->protocol->core, name, argument))) {
4130         pa_pstream_send_error(c->pstream, tag, PA_ERR_MODINITFAILED);
4131         return;
4132     }
4133
4134     reply = reply_new(tag);
4135     pa_tagstruct_putu32(reply, m->index);
4136     pa_pstream_send_tagstruct(c->pstream, reply);
4137 }
4138
4139 static void command_unload_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4140     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4141     uint32_t idx;
4142     pa_module *m;
4143
4144     pa_native_connection_assert_ref(c);
4145     pa_assert(t);
4146
4147     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4148         !pa_tagstruct_eof(t)) {
4149         protocol_error(c);
4150         return;
4151     }
4152
4153     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4154     m = pa_idxset_get_by_index(c->protocol->core->modules, idx);
4155     CHECK_VALIDITY(c->pstream, m, tag, PA_ERR_NOENTITY);
4156
4157     pa_module_unload_request(m, FALSE);
4158     pa_pstream_send_simple_ack(c->pstream, tag);
4159 }
4160
4161 static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4162     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4163     uint32_t idx = PA_INVALID_INDEX, idx_device = PA_INVALID_INDEX;
4164     const char *name_device = NULL;
4165
4166     pa_native_connection_assert_ref(c);
4167     pa_assert(t);
4168
4169     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4170         pa_tagstruct_getu32(t, &idx_device) < 0 ||
4171         pa_tagstruct_gets(t, &name_device) < 0 ||
4172         !pa_tagstruct_eof(t)) {
4173         protocol_error(c);
4174         return;
4175     }
4176
4177     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4178     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4179
4180     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);
4181     CHECK_VALIDITY(c->pstream, idx_device != PA_INVALID_INDEX || name_device, tag, PA_ERR_INVALID);
4182     CHECK_VALIDITY(c->pstream, idx_device == PA_INVALID_INDEX || !name_device, tag, PA_ERR_INVALID);
4183     CHECK_VALIDITY(c->pstream, !name_device || idx_device == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4184
4185     if (command == PA_COMMAND_MOVE_SINK_INPUT) {
4186         pa_sink_input *si = NULL;
4187         pa_sink *sink = NULL;
4188
4189         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
4190
4191         if (idx_device != PA_INVALID_INDEX)
4192             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx_device);
4193         else
4194             sink = pa_namereg_get(c->protocol->core, name_device, PA_NAMEREG_SINK);
4195
4196         CHECK_VALIDITY(c->pstream, si && sink, tag, PA_ERR_NOENTITY);
4197
4198         if (pa_sink_input_move_to(si, sink, TRUE) < 0) {
4199             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4200             return;
4201         }
4202     } else {
4203         pa_source_output *so = NULL;
4204         pa_source *source;
4205
4206         pa_assert(command == PA_COMMAND_MOVE_SOURCE_OUTPUT);
4207
4208         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
4209
4210         if (idx_device != PA_INVALID_INDEX)
4211             source = pa_idxset_get_by_index(c->protocol->core->sources, idx_device);
4212         else
4213             source = pa_namereg_get(c->protocol->core, name_device, PA_NAMEREG_SOURCE);
4214
4215         CHECK_VALIDITY(c->pstream, so && source, tag, PA_ERR_NOENTITY);
4216
4217         if (pa_source_output_move_to(so, source, TRUE) < 0) {
4218             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4219             return;
4220         }
4221     }
4222
4223     pa_pstream_send_simple_ack(c->pstream, tag);
4224 }
4225
4226 static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4227     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4228     uint32_t idx = PA_INVALID_INDEX;
4229     const char *name = NULL;
4230     pa_bool_t b;
4231
4232     pa_native_connection_assert_ref(c);
4233     pa_assert(t);
4234
4235     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4236         pa_tagstruct_gets(t, &name) < 0 ||
4237         pa_tagstruct_get_boolean(t, &b) < 0 ||
4238         !pa_tagstruct_eof(t)) {
4239         protocol_error(c);
4240         return;
4241     }
4242
4243     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4244     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);
4245     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4246     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4247     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4248
4249     if (command == PA_COMMAND_SUSPEND_SINK) {
4250
4251         if (idx == PA_INVALID_INDEX && name && !*name) {
4252
4253             pa_log_debug("%s all sinks", b ? "Suspending" : "Resuming");
4254
4255             if (pa_sink_suspend_all(c->protocol->core, b, PA_SUSPEND_USER) < 0) {
4256                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4257                 return;
4258             }
4259         } else {
4260             pa_sink *sink = NULL;
4261
4262             if (idx != PA_INVALID_INDEX)
4263                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
4264             else
4265                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
4266
4267             CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
4268
4269             if (pa_sink_suspend(sink, b, PA_SUSPEND_USER) < 0) {
4270                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4271                 return;
4272             }
4273         }
4274     } else {
4275
4276         pa_assert(command == PA_COMMAND_SUSPEND_SOURCE);
4277
4278         if (idx == PA_INVALID_INDEX && name && !*name) {
4279
4280             pa_log_debug("%s all sources", b ? "Suspending" : "Resuming");
4281
4282             if (pa_source_suspend_all(c->protocol->core, b, PA_SUSPEND_USER) < 0) {
4283                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4284                 return;
4285             }
4286
4287         } else {
4288             pa_source *source;
4289
4290             if (idx != PA_INVALID_INDEX)
4291                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
4292             else
4293                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
4294
4295             CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
4296
4297             if (pa_source_suspend(source, b, PA_SUSPEND_USER) < 0) {
4298                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
4299                 return;
4300             }
4301         }
4302     }
4303
4304     pa_pstream_send_simple_ack(c->pstream, tag);
4305 }
4306
4307 static void command_extension(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4308     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4309     uint32_t idx = PA_INVALID_INDEX;
4310     const char *name = NULL;
4311     pa_module *m;
4312     pa_native_protocol_ext_cb_t cb;
4313
4314     pa_native_connection_assert_ref(c);
4315     pa_assert(t);
4316
4317     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4318         pa_tagstruct_gets(t, &name) < 0) {
4319         protocol_error(c);
4320         return;
4321     }
4322
4323     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4324     CHECK_VALIDITY(c->pstream, !name || pa_utf8_valid(name), tag, PA_ERR_INVALID);
4325     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4326     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4327     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4328
4329     if (idx != PA_INVALID_INDEX)
4330         m = pa_idxset_get_by_index(c->protocol->core->modules, idx);
4331     else {
4332         for (m = pa_idxset_first(c->protocol->core->modules, &idx); m; m = pa_idxset_next(c->protocol->core->modules, &idx))
4333             if (strcmp(name, m->name) == 0)
4334                 break;
4335     }
4336
4337     CHECK_VALIDITY(c->pstream, m, tag, PA_ERR_NOEXTENSION);
4338     CHECK_VALIDITY(c->pstream, m->load_once || idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4339
4340     cb = (pa_native_protocol_ext_cb_t) (unsigned long) pa_hashmap_get(c->protocol->extensions, m);
4341     CHECK_VALIDITY(c->pstream, cb, tag, PA_ERR_NOEXTENSION);
4342
4343     if (cb(c->protocol, m, c, tag, t) < 0)
4344         protocol_error(c);
4345 }
4346
4347 static void command_set_card_profile(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4348     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4349     uint32_t idx = PA_INVALID_INDEX;
4350     const char *name = NULL, *profile = NULL;
4351     pa_card *card = NULL;
4352     int ret;
4353
4354     pa_native_connection_assert_ref(c);
4355     pa_assert(t);
4356
4357     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4358         pa_tagstruct_gets(t, &name) < 0 ||
4359         pa_tagstruct_gets(t, &profile) < 0 ||
4360         !pa_tagstruct_eof(t)) {
4361         protocol_error(c);
4362         return;
4363     }
4364
4365     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4366     CHECK_VALIDITY(c->pstream, !name || pa_namereg_is_valid_name(name), tag, PA_ERR_INVALID);
4367     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4368     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4369     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4370
4371     if (idx != PA_INVALID_INDEX)
4372         card = pa_idxset_get_by_index(c->protocol->core->cards, idx);
4373     else
4374         card = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_CARD);
4375
4376     CHECK_VALIDITY(c->pstream, card, tag, PA_ERR_NOENTITY);
4377
4378     if ((ret = pa_card_set_profile(card, profile, TRUE)) < 0) {
4379         pa_pstream_send_error(c->pstream, tag, -ret);
4380         return;
4381     }
4382
4383     pa_pstream_send_simple_ack(c->pstream, tag);
4384 }
4385
4386 static void command_set_sink_or_source_port(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
4387     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4388     uint32_t idx = PA_INVALID_INDEX;
4389     const char *name = NULL, *port = NULL;
4390     int ret;
4391
4392     pa_native_connection_assert_ref(c);
4393     pa_assert(t);
4394
4395     if (pa_tagstruct_getu32(t, &idx) < 0 ||
4396         pa_tagstruct_gets(t, &name) < 0 ||
4397         pa_tagstruct_gets(t, &port) < 0 ||
4398         !pa_tagstruct_eof(t)) {
4399         protocol_error(c);
4400         return;
4401     }
4402
4403     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
4404     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);
4405     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || name, tag, PA_ERR_INVALID);
4406     CHECK_VALIDITY(c->pstream, idx == PA_INVALID_INDEX || !name, tag, PA_ERR_INVALID);
4407     CHECK_VALIDITY(c->pstream, !name || idx == PA_INVALID_INDEX, tag, PA_ERR_INVALID);
4408
4409     if (command == PA_COMMAND_SET_SINK_PORT) {
4410         pa_sink *sink;
4411
4412         if (idx != PA_INVALID_INDEX)
4413             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
4414         else
4415             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK);
4416
4417         CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
4418
4419         if ((ret = pa_sink_set_port(sink, port, TRUE)) < 0) {
4420             pa_pstream_send_error(c->pstream, tag, -ret);
4421             return;
4422         }
4423     } else {
4424         pa_source *source;
4425
4426         pa_assert(command = PA_COMMAND_SET_SOURCE_PORT);
4427
4428         if (idx != PA_INVALID_INDEX)
4429             source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
4430         else
4431             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE);
4432
4433         CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
4434
4435         if ((ret = pa_source_set_port(source, port, TRUE)) < 0) {
4436             pa_pstream_send_error(c->pstream, tag, -ret);
4437             return;
4438         }
4439     }
4440
4441     pa_pstream_send_simple_ack(c->pstream, tag);
4442 }
4443
4444 /*** pstream callbacks ***/
4445
4446 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
4447     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4448
4449     pa_assert(p);
4450     pa_assert(packet);
4451     pa_native_connection_assert_ref(c);
4452
4453     if (pa_pdispatch_run(c->pdispatch, packet, creds, c) < 0) {
4454         pa_log("invalid packet.");
4455         native_connection_unlink(c);
4456     }
4457 }
4458
4459 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) {
4460     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4461     output_stream *stream;
4462
4463     pa_assert(p);
4464     pa_assert(chunk);
4465     pa_native_connection_assert_ref(c);
4466
4467     if (!(stream = OUTPUT_STREAM(pa_idxset_get_by_index(c->output_streams, channel)))) {
4468         pa_log_debug("Client sent block for invalid stream.");
4469         /* Ignoring */
4470         return;
4471     }
4472
4473 /*     pa_log("got %lu bytes", (unsigned long) chunk->length); */
4474
4475     if (playback_stream_isinstance(stream)) {
4476         playback_stream *ps = PLAYBACK_STREAM(stream);
4477
4478         pa_atomic_inc(&ps->seek_or_post_in_queue);
4479         if (chunk->memblock) {
4480             if (seek != PA_SEEK_RELATIVE || offset != 0)
4481                 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);
4482             else
4483                 pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
4484         } else
4485             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);
4486
4487     } else {
4488         upload_stream *u = UPLOAD_STREAM(stream);
4489         size_t l;
4490
4491         if (!u->memchunk.memblock) {
4492             if (u->length == chunk->length && chunk->memblock) {
4493                 u->memchunk = *chunk;
4494                 pa_memblock_ref(u->memchunk.memblock);
4495                 u->length = 0;
4496             } else {
4497                 u->memchunk.memblock = pa_memblock_new(c->protocol->core->mempool, u->length);
4498                 u->memchunk.index = u->memchunk.length = 0;
4499             }
4500         }
4501
4502         pa_assert(u->memchunk.memblock);
4503
4504         l = u->length;
4505         if (l > chunk->length)
4506             l = chunk->length;
4507
4508         if (l > 0) {
4509             void *dst;
4510             dst = pa_memblock_acquire(u->memchunk.memblock);
4511
4512             if (chunk->memblock) {
4513                 void *src;
4514                 src = pa_memblock_acquire(chunk->memblock);
4515
4516                 memcpy((uint8_t*) dst + u->memchunk.index + u->memchunk.length,
4517                        (uint8_t*) src + chunk->index, l);
4518
4519                 pa_memblock_release(chunk->memblock);
4520             } else
4521                 pa_silence_memory((uint8_t*) dst + u->memchunk.index + u->memchunk.length, l, &u->sample_spec);
4522
4523             pa_memblock_release(u->memchunk.memblock);
4524
4525             u->memchunk.length += l;
4526             u->length -= l;
4527         }
4528     }
4529 }
4530
4531 static void pstream_die_callback(pa_pstream *p, void *userdata) {
4532     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4533
4534     pa_assert(p);
4535     pa_native_connection_assert_ref(c);
4536
4537     native_connection_unlink(c);
4538     pa_log_info("Connection died.");
4539 }
4540
4541 static void pstream_drain_callback(pa_pstream *p, void *userdata) {
4542     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4543
4544     pa_assert(p);
4545     pa_native_connection_assert_ref(c);
4546
4547     native_connection_send_memblock(c);
4548 }
4549
4550 static void pstream_revoke_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
4551     pa_thread_mq *q;
4552
4553     if (!(q = pa_thread_mq_get()))
4554         pa_pstream_send_revoke(p, block_id);
4555     else
4556         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_REVOKE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
4557 }
4558
4559 static void pstream_release_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
4560     pa_thread_mq *q;
4561
4562     if (!(q = pa_thread_mq_get()))
4563         pa_pstream_send_release(p, block_id);
4564     else
4565         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_RELEASE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
4566 }
4567
4568 /*** client callbacks ***/
4569
4570 static void client_kill_cb(pa_client *c) {
4571     pa_assert(c);
4572
4573     native_connection_unlink(PA_NATIVE_CONNECTION(c->userdata));
4574     pa_log_info("Connection killed.");
4575 }
4576
4577 static void client_send_event_cb(pa_client *client, const char*event, pa_proplist *pl) {
4578     pa_tagstruct *t;
4579     pa_native_connection *c;
4580
4581     pa_assert(client);
4582     c = PA_NATIVE_CONNECTION(client->userdata);
4583     pa_native_connection_assert_ref(c);
4584
4585     if (c->version < 15)
4586       return;
4587
4588     t = pa_tagstruct_new(NULL, 0);
4589     pa_tagstruct_putu32(t, PA_COMMAND_CLIENT_EVENT);
4590     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
4591     pa_tagstruct_puts(t, event);
4592     pa_tagstruct_put_proplist(t, pl);
4593     pa_pstream_send_tagstruct(c->pstream, t);
4594 }
4595
4596 /*** module entry points ***/
4597
4598 static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *t, void *userdata) {
4599     pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
4600
4601     pa_assert(m);
4602     pa_native_connection_assert_ref(c);
4603     pa_assert(c->auth_timeout_event == e);
4604
4605     if (!c->authorized) {
4606         native_connection_unlink(c);
4607         pa_log_info("Connection terminated due to authentication timeout.");
4608     }
4609 }
4610
4611 void pa_native_protocol_connect(pa_native_protocol *p, pa_iochannel *io, pa_native_options *o) {
4612     pa_native_connection *c;
4613     char pname[128];
4614     pa_client *client;
4615     pa_client_new_data data;
4616
4617     pa_assert(p);
4618     pa_assert(io);
4619     pa_assert(o);
4620
4621     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
4622         pa_log_warn("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
4623         pa_iochannel_free(io);
4624         return;
4625     }
4626
4627     pa_client_new_data_init(&data);
4628     data.module = o->module;
4629     data.driver = __FILE__;
4630     pa_iochannel_socket_peer_to_string(io, pname, sizeof(pname));
4631     pa_proplist_setf(data.proplist, PA_PROP_APPLICATION_NAME, "Native client (%s)", pname);
4632     pa_proplist_sets(data.proplist, "native-protocol.peer", pname);
4633     client = pa_client_new(p->core, &data);
4634     pa_client_new_data_done(&data);
4635
4636     if (!client)
4637         return;
4638
4639     c = pa_msgobject_new(pa_native_connection);
4640     c->parent.parent.free = native_connection_free;
4641     c->parent.process_msg = native_connection_process_msg;
4642     c->protocol = p;
4643     c->options = pa_native_options_ref(o);
4644     c->authorized = FALSE;
4645
4646     if (o->auth_anonymous) {
4647         pa_log_info("Client authenticated anonymously.");
4648         c->authorized = TRUE;
4649     }
4650
4651     if (!c->authorized &&
4652         o->auth_ip_acl &&
4653         pa_ip_acl_check(o->auth_ip_acl, pa_iochannel_get_recv_fd(io)) > 0) {
4654
4655         pa_log_info("Client authenticated by IP ACL.");
4656         c->authorized = TRUE;
4657     }
4658
4659     if (!c->authorized)
4660         c->auth_timeout_event = pa_core_rttime_new(p->core, pa_rtclock_now() + AUTH_TIMEOUT, auth_timeout, c);
4661     else
4662         c->auth_timeout_event = NULL;
4663
4664     c->is_local = pa_iochannel_socket_is_local(io);
4665     c->version = 8;
4666
4667     c->client = client;
4668     c->client->kill = client_kill_cb;
4669     c->client->send_event = client_send_event_cb;
4670     c->client->userdata = c;
4671
4672     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->mempool);
4673     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
4674     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
4675     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
4676     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
4677     pa_pstream_set_revoke_callback(c->pstream, pstream_revoke_callback, c);
4678     pa_pstream_set_release_callback(c->pstream, pstream_release_callback, c);
4679
4680     c->pdispatch = pa_pdispatch_new(p->core->mainloop, TRUE, command_table, PA_COMMAND_MAX);
4681
4682     c->record_streams = pa_idxset_new(NULL, NULL);
4683     c->output_streams = pa_idxset_new(NULL, NULL);
4684
4685     c->rrobin_index = PA_IDXSET_INVALID;
4686     c->subscription = NULL;
4687
4688     pa_idxset_put(p->connections, c, NULL);
4689
4690 #ifdef HAVE_CREDS
4691     if (pa_iochannel_creds_supported(io))
4692         pa_iochannel_creds_enable(io);
4693 #endif
4694
4695     pa_hook_fire(&p->hooks[PA_NATIVE_HOOK_CONNECTION_PUT], c);
4696 }
4697
4698 void pa_native_protocol_disconnect(pa_native_protocol *p, pa_module *m) {
4699     pa_native_connection *c;
4700     void *state = NULL;
4701
4702     pa_assert(p);
4703     pa_assert(m);
4704
4705     while ((c = pa_idxset_iterate(p->connections, &state, NULL)))
4706         if (c->options->module == m)
4707             native_connection_unlink(c);
4708 }
4709
4710 static pa_native_protocol* native_protocol_new(pa_core *c) {
4711     pa_native_protocol *p;
4712     pa_native_hook_t h;
4713
4714     pa_assert(c);
4715
4716     p = pa_xnew(pa_native_protocol, 1);
4717     PA_REFCNT_INIT(p);
4718     p->core = c;
4719     p->connections = pa_idxset_new(NULL, NULL);
4720
4721     p->servers = NULL;
4722
4723     p->extensions = pa_hashmap_new(pa_idxset_trivial_hash_func, pa_idxset_trivial_compare_func);
4724
4725     for (h = 0; h < PA_NATIVE_HOOK_MAX; h++)
4726         pa_hook_init(&p->hooks[h], p);
4727
4728     pa_assert_se(pa_shared_set(c, "native-protocol", p) >= 0);
4729
4730     return p;
4731 }
4732
4733 pa_native_protocol* pa_native_protocol_get(pa_core *c) {
4734     pa_native_protocol *p;
4735
4736     if ((p = pa_shared_get(c, "native-protocol")))
4737         return pa_native_protocol_ref(p);
4738
4739     return native_protocol_new(c);
4740 }
4741
4742 pa_native_protocol* pa_native_protocol_ref(pa_native_protocol *p) {
4743     pa_assert(p);
4744     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4745
4746     PA_REFCNT_INC(p);
4747
4748     return p;
4749 }
4750
4751 void pa_native_protocol_unref(pa_native_protocol *p) {
4752     pa_native_connection *c;
4753     pa_native_hook_t h;
4754
4755     pa_assert(p);
4756     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4757
4758     if (PA_REFCNT_DEC(p) > 0)
4759         return;
4760
4761     while ((c = pa_idxset_first(p->connections, NULL)))
4762         native_connection_unlink(c);
4763
4764     pa_idxset_free(p->connections, NULL, NULL);
4765
4766     pa_strlist_free(p->servers);
4767
4768     for (h = 0; h < PA_NATIVE_HOOK_MAX; h++)
4769         pa_hook_done(&p->hooks[h]);
4770
4771     pa_hashmap_free(p->extensions, NULL, NULL);
4772
4773     pa_assert_se(pa_shared_remove(p->core, "native-protocol") >= 0);
4774
4775     pa_xfree(p);
4776 }
4777
4778 void pa_native_protocol_add_server_string(pa_native_protocol *p, const char *name) {
4779     pa_assert(p);
4780     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4781     pa_assert(name);
4782
4783     p->servers = pa_strlist_prepend(p->servers, name);
4784
4785     pa_hook_fire(&p->hooks[PA_NATIVE_HOOK_SERVERS_CHANGED], p->servers);
4786 }
4787
4788 void pa_native_protocol_remove_server_string(pa_native_protocol *p, const char *name) {
4789     pa_assert(p);
4790     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4791     pa_assert(name);
4792
4793     p->servers = pa_strlist_remove(p->servers, name);
4794
4795     pa_hook_fire(&p->hooks[PA_NATIVE_HOOK_SERVERS_CHANGED], p->servers);
4796 }
4797
4798 pa_hook *pa_native_protocol_hooks(pa_native_protocol *p) {
4799     pa_assert(p);
4800     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4801
4802     return p->hooks;
4803 }
4804
4805 pa_strlist *pa_native_protocol_servers(pa_native_protocol *p) {
4806     pa_assert(p);
4807     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4808
4809     return p->servers;
4810 }
4811
4812 int pa_native_protocol_install_ext(pa_native_protocol *p, pa_module *m, pa_native_protocol_ext_cb_t cb) {
4813     pa_assert(p);
4814     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4815     pa_assert(m);
4816     pa_assert(cb);
4817     pa_assert(!pa_hashmap_get(p->extensions, m));
4818
4819     pa_assert_se(pa_hashmap_put(p->extensions, m, (void*) (unsigned long) cb) == 0);
4820     return 0;
4821 }
4822
4823 void pa_native_protocol_remove_ext(pa_native_protocol *p, pa_module *m) {
4824     pa_assert(p);
4825     pa_assert(PA_REFCNT_VALUE(p) >= 1);
4826     pa_assert(m);
4827
4828     pa_assert_se(pa_hashmap_remove(p->extensions, m));
4829 }
4830
4831 pa_native_options* pa_native_options_new(void) {
4832     pa_native_options *o;
4833
4834     o = pa_xnew0(pa_native_options, 1);
4835     PA_REFCNT_INIT(o);
4836
4837     return o;
4838 }
4839
4840 pa_native_options* pa_native_options_ref(pa_native_options *o) {
4841     pa_assert(o);
4842     pa_assert(PA_REFCNT_VALUE(o) >= 1);
4843
4844     PA_REFCNT_INC(o);
4845
4846     return o;
4847 }
4848
4849 void pa_native_options_unref(pa_native_options *o) {
4850     pa_assert(o);
4851     pa_assert(PA_REFCNT_VALUE(o) >= 1);
4852
4853     if (PA_REFCNT_DEC(o) > 0)
4854         return;
4855
4856     pa_xfree(o->auth_group);
4857
4858     if (o->auth_ip_acl)
4859         pa_ip_acl_free(o->auth_ip_acl);
4860
4861     if (o->auth_cookie)
4862         pa_auth_cookie_unref(o->auth_cookie);
4863
4864     pa_xfree(o);
4865 }
4866
4867 int pa_native_options_parse(pa_native_options *o, pa_core *c, pa_modargs *ma) {
4868     pa_bool_t enabled;
4869     const char *acl;
4870
4871     pa_assert(o);
4872     pa_assert(PA_REFCNT_VALUE(o) >= 1);
4873     pa_assert(ma);
4874
4875     if (pa_modargs_get_value_boolean(ma, "auth-anonymous", &o->auth_anonymous) < 0) {
4876         pa_log("auth-anonymous= expects a boolean argument.");
4877         return -1;
4878     }
4879
4880     enabled = TRUE;
4881     if (pa_modargs_get_value_boolean(ma, "auth-group-enabled", &enabled) < 0) {
4882         pa_log("auth-group-enabled= expects a boolean argument.");
4883         return -1;
4884     }
4885
4886     pa_xfree(o->auth_group);
4887     o->auth_group = enabled ? pa_xstrdup(pa_modargs_get_value(ma, "auth-group", pa_in_system_mode() ? PA_ACCESS_GROUP : NULL)) : NULL;
4888
4889 #ifndef HAVE_CREDS
4890     if (o->auth_group)
4891         pa_log_warn("Authentication group configured, but not available on local system. Ignoring.");
4892 #endif
4893
4894     if ((acl = pa_modargs_get_value(ma, "auth-ip-acl", NULL))) {
4895         pa_ip_acl *ipa;
4896
4897         if (!(ipa = pa_ip_acl_new(acl))) {
4898             pa_log("Failed to parse IP ACL '%s'", acl);
4899             return -1;
4900         }
4901
4902         if (o->auth_ip_acl)
4903             pa_ip_acl_free(o->auth_ip_acl);
4904
4905         o->auth_ip_acl = ipa;
4906     }
4907
4908     enabled = TRUE;
4909     if (pa_modargs_get_value_boolean(ma, "auth-cookie-enabled", &enabled) < 0) {
4910         pa_log("auth-cookie-enabled= expects a boolean argument.");
4911         return -1;
4912     }
4913
4914     if (o->auth_cookie)
4915         pa_auth_cookie_unref(o->auth_cookie);
4916
4917     if (enabled) {
4918         const char *cn;
4919
4920         /* The new name for this is 'auth-cookie', for compat reasons
4921          * we check the old name too */
4922         if (!(cn = pa_modargs_get_value(ma, "auth-cookie", NULL)))
4923             if (!(cn = pa_modargs_get_value(ma, "cookie", NULL)))
4924                 cn = PA_NATIVE_COOKIE_FILE;
4925
4926         if (!(o->auth_cookie = pa_auth_cookie_get(c, cn, PA_NATIVE_COOKIE_LENGTH)))
4927             return -1;
4928
4929     } else
4930           o->auth_cookie = NULL;
4931
4932     return 0;
4933 }
4934
4935 pa_pstream* pa_native_connection_get_pstream(pa_native_connection *c) {
4936     pa_native_connection_assert_ref(c);
4937
4938     return c->pstream;
4939 }
4940
4941 pa_client* pa_native_connection_get_client(pa_native_connection *c) {
4942     pa_native_connection_assert_ref(c);
4943
4944     return c->client;
4945 }