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