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