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