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