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