code.delx.au - pulseaudio/blob - src/pulsecore/protocol-native.c

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