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