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

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2004-2006 Lennart Poettering
   7   Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
   8
   9   PulseAudio is free software; you can redistribute it and/or modify
  10   it under the terms of the GNU Lesser General Public License as published
  11   by the Free Software Foundation; either version 2 of the License,
  12   or (at your option) any later version.
  13
  14   PulseAudio is distributed in the hope that it will be useful, but
  15   WITHOUT ANY WARRANTY; without even the implied warranty of
  16   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  17   General Public License for more details.
  18
  19   You should have received a copy of the GNU Lesser General Public License
  20   along with PulseAudio; if not, write to the Free Software
  21   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  22   USA.
  23 ***/
  24
  25 #ifdef HAVE_CONFIG_H
  26 #include <config.h>
  27 #endif
  28
  29 #include <string.h>
  30 #include <stdio.h>
  31 #include <stdlib.h>
  32 #include <unistd.h>
  33
  34 #include <pulse/timeval.h>
  35 #include <pulse/version.h>
  36 #include <pulse/utf8.h>
  37 #include <pulse/util.h>
  38 #include <pulse/xmalloc.h>
  39
  40 #include <pulsecore/native-common.h>
  41 #include <pulsecore/packet.h>
  42 #include <pulsecore/client.h>
  43 #include <pulsecore/source-output.h>
  44 #include <pulsecore/sink-input.h>
  45 #include <pulsecore/pstream.h>
  46 #include <pulsecore/tagstruct.h>
  47 #include <pulsecore/pdispatch.h>
  48 #include <pulsecore/pstream-util.h>
  49 #include <pulsecore/authkey.h>
  50 #include <pulsecore/namereg.h>
  51 #include <pulsecore/core-scache.h>
  52 #include <pulsecore/core-subscribe.h>
  53 #include <pulsecore/log.h>
  54 #include <pulsecore/autoload.h>
  55 #include <pulsecore/authkey-prop.h>
  56 #include <pulsecore/strlist.h>
  57 #include <pulsecore/props.h>
  58 #include <pulsecore/sample-util.h>
  59 #include <pulsecore/llist.h>
  60 #include <pulsecore/creds.h>
  61 #include <pulsecore/core-util.h>
  62 #include <pulsecore/ipacl.h>
  63 #include <pulsecore/thread-mq.h>
  64
  65 #include "protocol-native.h"
  66
  67 /* Kick a client if it doesn't authenticate within this time */
  68 #define AUTH_TIMEOUT 60
  69
  70 /* Don't accept more connection than this */
  71 #define MAX_CONNECTIONS 64
  72
  73 #define MAX_MEMBLOCKQ_LENGTH (4*1024*1024) /* 4MB */
  74
  75 typedef struct connection connection;
  76 struct pa_protocol_native;
  77
  78 typedef struct record_stream {
  79     pa_msgobject parent;
  80
  81     connection *connection;
  82     uint32_t index;
  83
  84     pa_source_output *source_output;
  85     pa_memblockq *memblockq;
  86     size_t fragment_size;
  87 } record_stream;
  88
  89 typedef struct output_stream {
  90     pa_msgobject parent;
  91 } output_stream;
  92
  93 typedef struct playback_stream {
  94     output_stream parent;
  95
  96     connection *connection;
  97     uint32_t index;
  98
  99     pa_sink_input *sink_input;
 100     pa_memblockq *memblockq;
 101     int drain_request;
 102     uint32_t drain_tag;
 103     uint32_t syncid;
 104     int underrun;
 105
 106     pa_atomic_t missing;
 107     size_t minreq;
 108
 109     /* Only updated after SINK_INPUT_MESSAGE_UPDATE_LATENCY */
 110     int64_t read_index, write_index;
 111     size_t resampled_chunk_length;
 112 } playback_stream;
 113
 114 typedef struct upload_stream {
 115     output_stream parent;
 116
 117     connection *connection;
 118     uint32_t index;
 119
 120     pa_memchunk memchunk;
 121     size_t length;
 122     char *name;
 123     pa_sample_spec sample_spec;
 124     pa_channel_map channel_map;
 125 } upload_stream;
 126
 127 struct connection {
 128     pa_msgobject parent;
 129
 130     int authorized;
 131     uint32_t version;
 132     pa_protocol_native *protocol;
 133     pa_client *client;
 134     pa_pstream *pstream;
 135     pa_pdispatch *pdispatch;
 136     pa_idxset *record_streams, *output_streams;
 137     uint32_t rrobin_index;
 138     pa_subscription *subscription;
 139     pa_time_event *auth_timeout_event;
 140 };
 141
 142 PA_DECLARE_CLASS(record_stream);
 143 #define RECORD_STREAM(o) (record_stream_cast(o))
 144 static PA_DEFINE_CHECK_TYPE(record_stream, pa_msgobject);
 145
 146 PA_DECLARE_CLASS(output_stream);
 147 #define OUTPUT_STREAM(o) (output_stream_cast(o))
 148 static PA_DEFINE_CHECK_TYPE(output_stream, pa_msgobject);
 149
 150 PA_DECLARE_CLASS(playback_stream);
 151 #define PLAYBACK_STREAM(o) (playback_stream_cast(o))
 152 static PA_DEFINE_CHECK_TYPE(playback_stream, output_stream);
 153
 154 PA_DECLARE_CLASS(upload_stream);
 155 #define UPLOAD_STREAM(o) (upload_stream_cast(o))
 156 static PA_DEFINE_CHECK_TYPE(upload_stream, output_stream);
 157
 158 PA_DECLARE_CLASS(connection);
 159 #define CONNECTION(o) (connection_cast(o))
 160 static PA_DEFINE_CHECK_TYPE(connection, pa_msgobject);
 161
 162 struct pa_protocol_native {
 163     pa_module *module;
 164     pa_core *core;
 165     int public;
 166     pa_socket_server *server;
 167     pa_idxset *connections;
 168     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
 169     int auth_cookie_in_property;
 170 #ifdef HAVE_CREDS
 171     char *auth_group;
 172 #endif
 173     pa_ip_acl *auth_ip_acl;
 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 };
 192
 193 enum {
 194     RECORD_STREAM_MESSAGE_POST_DATA         /* data from source output to main loop */
 195 };
 196
 197 enum {
 198     CONNECTION_MESSAGE_RELEASE,
 199     CONNECTION_MESSAGE_REVOKE
 200 };
 201
 202 static int sink_input_peek_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk);
 203 static void sink_input_drop_cb(pa_sink_input *i, size_t length);
 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_moved_cb(pa_sink_input *i);
 207
 208 static void send_memblock(connection *c);
 209 static void request_bytes(struct playback_stream*s);
 210
 211 static void source_output_kill_cb(pa_source_output *o);
 212 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk);
 213 static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend);
 214 static void source_output_moved_cb(pa_source_output *o);
 215 static pa_usec_t source_output_get_latency_cb(pa_source_output *o);
 216
 217 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
 218
 219 static void command_exit(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 220 static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 221 static void command_drain_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 222 static void command_create_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 223 static void command_delete_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 224 static void command_auth(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 225 static void command_set_client_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 226 static void command_lookup(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 227 static void command_stat(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 228 static void command_get_playback_latency(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 229 static void command_get_record_latency(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 230 static void command_create_upload_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 231 static void command_finish_upload_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 232 static void command_play_sample(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 233 static void command_remove_sample(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 234 static void command_get_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 235 static void command_get_info_list(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 236 static void command_get_server_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 237 static void command_subscribe(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 238 static void command_set_volume(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 239 static void command_set_mute(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 240 static void command_cork_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 241 static void command_trigger_or_flush_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 242 static void command_set_default_sink_or_source(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 243 static void command_set_stream_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 244 static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 245 static void command_load_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 246 static void command_unload_module(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 247 static void command_add_autoload(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 248 static void command_remove_autoload(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 249 static void command_get_autoload_info(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 250 static void command_get_autoload_info_list(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 251 static void command_cork_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 252 static void command_flush_record_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 253 static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 254 static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 255 static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 256 static void command_update_stream_sample_rate(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 257
 258 static const pa_pdispatch_cb_t command_table[PA_COMMAND_MAX] = {
 259     [PA_COMMAND_ERROR] = NULL,
 260     [PA_COMMAND_TIMEOUT] = NULL,
 261     [PA_COMMAND_REPLY] = NULL,
 262     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = command_create_playback_stream,
 263     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = command_delete_stream,
 264     [PA_COMMAND_DRAIN_PLAYBACK_STREAM] = command_drain_playback_stream,
 265     [PA_COMMAND_CREATE_RECORD_STREAM] = command_create_record_stream,
 266     [PA_COMMAND_DELETE_RECORD_STREAM] = command_delete_stream,
 267     [PA_COMMAND_AUTH] = command_auth,
 268     [PA_COMMAND_REQUEST] = NULL,
 269     [PA_COMMAND_EXIT] = command_exit,
 270     [PA_COMMAND_SET_CLIENT_NAME] = command_set_client_name,
 271     [PA_COMMAND_LOOKUP_SINK] = command_lookup,
 272     [PA_COMMAND_LOOKUP_SOURCE] = command_lookup,
 273     [PA_COMMAND_STAT] = command_stat,
 274     [PA_COMMAND_GET_PLAYBACK_LATENCY] = command_get_playback_latency,
 275     [PA_COMMAND_GET_RECORD_LATENCY] = command_get_record_latency,
 276     [PA_COMMAND_CREATE_UPLOAD_STREAM] = command_create_upload_stream,
 277     [PA_COMMAND_DELETE_UPLOAD_STREAM] = command_delete_stream,
 278     [PA_COMMAND_FINISH_UPLOAD_STREAM] = command_finish_upload_stream,
 279     [PA_COMMAND_PLAY_SAMPLE] = command_play_sample,
 280     [PA_COMMAND_REMOVE_SAMPLE] = command_remove_sample,
 281     [PA_COMMAND_GET_SINK_INFO] = command_get_info,
 282     [PA_COMMAND_GET_SOURCE_INFO] = command_get_info,
 283     [PA_COMMAND_GET_CLIENT_INFO] = command_get_info,
 284     [PA_COMMAND_GET_MODULE_INFO] = command_get_info,
 285     [PA_COMMAND_GET_SINK_INPUT_INFO] = command_get_info,
 286     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO] = command_get_info,
 287     [PA_COMMAND_GET_SAMPLE_INFO] = command_get_info,
 288     [PA_COMMAND_GET_SINK_INFO_LIST] = command_get_info_list,
 289     [PA_COMMAND_GET_SOURCE_INFO_LIST] = command_get_info_list,
 290     [PA_COMMAND_GET_MODULE_INFO_LIST] = command_get_info_list,
 291     [PA_COMMAND_GET_CLIENT_INFO_LIST] = command_get_info_list,
 292     [PA_COMMAND_GET_SINK_INPUT_INFO_LIST] = command_get_info_list,
 293     [PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST] = command_get_info_list,
 294     [PA_COMMAND_GET_SAMPLE_INFO_LIST] = command_get_info_list,
 295     [PA_COMMAND_GET_SERVER_INFO] = command_get_server_info,
 296     [PA_COMMAND_SUBSCRIBE] = command_subscribe,
 297
 298     [PA_COMMAND_SET_SINK_VOLUME] = command_set_volume,
 299     [PA_COMMAND_SET_SINK_INPUT_VOLUME] = command_set_volume,
 300     [PA_COMMAND_SET_SOURCE_VOLUME] = command_set_volume,
 301
 302     [PA_COMMAND_SET_SINK_MUTE] = command_set_mute,
 303     [PA_COMMAND_SET_SINK_INPUT_MUTE] = command_set_mute,
 304     [PA_COMMAND_SET_SOURCE_MUTE] = command_set_mute,
 305
 306     [PA_COMMAND_SUSPEND_SINK] = command_suspend,
 307     [PA_COMMAND_SUSPEND_SOURCE] = command_suspend,
 308
 309     [PA_COMMAND_CORK_PLAYBACK_STREAM] = command_cork_playback_stream,
 310     [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
 311     [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
 312     [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
 313
 314     [PA_COMMAND_CORK_RECORD_STREAM] = command_cork_record_stream,
 315     [PA_COMMAND_FLUSH_RECORD_STREAM] = command_flush_record_stream,
 316
 317     [PA_COMMAND_SET_DEFAULT_SINK] = command_set_default_sink_or_source,
 318     [PA_COMMAND_SET_DEFAULT_SOURCE] = command_set_default_sink_or_source,
 319     [PA_COMMAND_SET_PLAYBACK_STREAM_NAME] = command_set_stream_name,
 320     [PA_COMMAND_SET_RECORD_STREAM_NAME] = command_set_stream_name,
 321     [PA_COMMAND_KILL_CLIENT] = command_kill,
 322     [PA_COMMAND_KILL_SINK_INPUT] = command_kill,
 323     [PA_COMMAND_KILL_SOURCE_OUTPUT] = command_kill,
 324     [PA_COMMAND_LOAD_MODULE] = command_load_module,
 325     [PA_COMMAND_UNLOAD_MODULE] = command_unload_module,
 326     [PA_COMMAND_GET_AUTOLOAD_INFO] = command_get_autoload_info,
 327     [PA_COMMAND_GET_AUTOLOAD_INFO_LIST] = command_get_autoload_info_list,
 328     [PA_COMMAND_ADD_AUTOLOAD] = command_add_autoload,
 329     [PA_COMMAND_REMOVE_AUTOLOAD] = command_remove_autoload,
 330
 331     [PA_COMMAND_MOVE_SINK_INPUT] = command_move_stream,
 332     [PA_COMMAND_MOVE_SOURCE_OUTPUT] = command_move_stream,
 333
 334     [PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR] = command_set_stream_buffer_attr,
 335     [PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR] = command_set_stream_buffer_attr,
 336
 337     [PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE] = command_update_stream_sample_rate,
 338     [PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE] = command_update_stream_sample_rate
 339 };
 340
 341 /* structure management */
 342
 343 static void upload_stream_unlink(upload_stream *s) {
 344     pa_assert(s);
 345
 346     if (!s->connection)
 347         return;
 348
 349     pa_assert_se(pa_idxset_remove_by_data(s->connection->output_streams, s, NULL) == s);
 350     s->connection = NULL;
 351     upload_stream_unref(s);
 352 }
 353
 354 static void upload_stream_free(pa_object *o) {
 355     upload_stream *s = UPLOAD_STREAM(o);
 356     pa_assert(s);
 357
 358     upload_stream_unlink(s);
 359
 360     pa_xfree(s->name);
 361
 362     if (s->memchunk.memblock)
 363         pa_memblock_unref(s->memchunk.memblock);
 364
 365     pa_xfree(s);
 366 }
 367
 368 static upload_stream* upload_stream_new(
 369         connection *c,
 370         const pa_sample_spec *ss,
 371         const pa_channel_map *map,
 372         const char *name, size_t length) {
 373
 374     upload_stream *s;
 375
 376     pa_assert(c);
 377     pa_assert(ss);
 378     pa_assert(name);
 379     pa_assert(length > 0);
 380
 381     s = pa_msgobject_new(upload_stream);
 382     s->parent.parent.parent.free = upload_stream_free;
 383     s->connection = c;
 384     s->sample_spec = *ss;
 385     s->channel_map = *map;
 386     s->name = pa_xstrdup(name);
 387     pa_memchunk_reset(&s->memchunk);
 388     s->length = length;
 389
 390     pa_idxset_put(c->output_streams, s, &s->index);
 391
 392     return s;
 393 }
 394
 395 static void record_stream_unlink(record_stream *s) {
 396     pa_assert(s);
 397
 398     if (!s->connection)
 399         return;
 400
 401     if (s->source_output) {
 402         pa_source_output_unlink(s->source_output);
 403         pa_source_output_unref(s->source_output);
 404         s->source_output = NULL;
 405     }
 406
 407     pa_assert_se(pa_idxset_remove_by_data(s->connection->record_streams, s, NULL) == s);
 408     s->connection = NULL;
 409     record_stream_unref(s);
 410 }
 411
 412 static void record_stream_free(pa_object *o) {
 413     record_stream *s = RECORD_STREAM(o);
 414     pa_assert(s);
 415
 416     record_stream_unlink(s);
 417
 418     pa_memblockq_free(s->memblockq);
 419     pa_xfree(s);
 420 }
 421
 422 static int record_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 423     record_stream *s = RECORD_STREAM(o);
 424     record_stream_assert_ref(s);
 425
 426     if (!s->connection)
 427         return -1;
 428
 429     switch (code) {
 430
 431         case RECORD_STREAM_MESSAGE_POST_DATA:
 432
 433             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
 434 /*                 pa_log_warn("Failed to push data into output queue."); */
 435                 return -1;
 436             }
 437
 438             if (!pa_pstream_is_pending(s->connection->pstream))
 439                 send_memblock(s->connection);
 440
 441             break;
 442     }
 443
 444     return 0;
 445 }
 446
 447 static record_stream* record_stream_new(
 448         connection *c,
 449         pa_source *source,
 450         pa_sample_spec *ss,
 451         pa_channel_map *map,
 452         const char *name,
 453         uint32_t *maxlength,
 454         uint32_t fragment_size,
 455         pa_source_output_flags_t flags) {
 456
 457     record_stream *s;
 458     pa_source_output *source_output;
 459     size_t base;
 460     pa_source_output_new_data data;
 461
 462     pa_assert(c);
 463     pa_assert(ss);
 464     pa_assert(name);
 465     pa_assert(maxlength);
 466     pa_assert(*maxlength > 0);
 467
 468     pa_source_output_new_data_init(&data);
 469     data.module = c->protocol->module;
 470     data.client = c->client;
 471     data.source = source;
 472     data.driver = __FILE__;
 473     data.name = name;
 474     pa_source_output_new_data_set_sample_spec(&data, ss);
 475     pa_source_output_new_data_set_channel_map(&data, map);
 476
 477     if (!(source_output = pa_source_output_new(c->protocol->core, &data, flags)))
 478         return NULL;
 479
 480     s = pa_msgobject_new(record_stream);
 481     s->parent.parent.free = record_stream_free;
 482     s->parent.process_msg = record_stream_process_msg;
 483     s->connection = c;
 484     s->source_output = source_output;
 485     s->source_output->push = source_output_push_cb;
 486     s->source_output->kill = source_output_kill_cb;
 487     s->source_output->get_latency = source_output_get_latency_cb;
 488     s->source_output->moved = source_output_moved_cb;
 489     s->source_output->suspend = source_output_suspend_cb;
 490     s->source_output->userdata = s;
 491
 492     s->memblockq = pa_memblockq_new(
 493             0,
 494             *maxlength,
 495             0,
 496             base = pa_frame_size(&s->source_output->sample_spec),
 497             1,
 498             0,
 499             NULL);
 500
 501     *maxlength = pa_memblockq_get_maxlength(s->memblockq);
 502
 503     s->fragment_size = (fragment_size/base)*base;
 504     if (s->fragment_size <= 0)
 505         s->fragment_size = base;
 506
 507     if (s->fragment_size > *maxlength)
 508         s->fragment_size = *maxlength;
 509
 510     *ss = s->source_output->sample_spec;
 511     *map = s->source_output->channel_map;
 512
 513     pa_idxset_put(c->record_streams, s, &s->index);
 514
 515     pa_source_output_put(s->source_output);
 516     return s;
 517 }
 518
 519 static void playback_stream_unlink(playback_stream *s) {
 520     pa_assert(s);
 521
 522     if (!s->connection)
 523         return;
 524
 525     if (s->sink_input) {
 526         pa_sink_input_unlink(s->sink_input);
 527         pa_sink_input_unref(s->sink_input);
 528         s->sink_input = NULL;
 529     }
 530
 531     if (s->drain_request)
 532         pa_pstream_send_error(s->connection->pstream, s->drain_tag, PA_ERR_NOENTITY);
 533
 534     pa_assert_se(pa_idxset_remove_by_data(s->connection->output_streams, s, NULL) == s);
 535     s->connection = NULL;
 536     playback_stream_unref(s);
 537 }
 538
 539 static void playback_stream_free(pa_object* o) {
 540     playback_stream *s = PLAYBACK_STREAM(o);
 541     pa_assert(s);
 542
 543     playback_stream_unlink(s);
 544
 545     pa_memblockq_free(s->memblockq);
 546     pa_xfree(s);
 547 }
 548
 549 static int playback_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 550     playback_stream *s = PLAYBACK_STREAM(o);
 551     playback_stream_assert_ref(s);
 552
 553     if (!s->connection)
 554         return -1;
 555
 556     switch (code) {
 557         case PLAYBACK_STREAM_MESSAGE_REQUEST_DATA: {
 558             pa_tagstruct *t;
 559             uint32_t l = 0;
 560
 561             for (;;) {
 562                 int32_t k;
 563
 564                 if ((k = pa_atomic_load(&s->missing)) <= 0)
 565                     break;
 566
 567                 l += k;
 568
 569                 if (l < s->minreq)
 570                     break;
 571
 572                 if (pa_atomic_sub(&s->missing, k) <= k)
 573                     break;
 574             }
 575
 576             if (l < s->minreq)
 577                 break;
 578
 579             t = pa_tagstruct_new(NULL, 0);
 580             pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
 581             pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 582             pa_tagstruct_putu32(t, s->index);
 583             pa_tagstruct_putu32(t, l);
 584             pa_pstream_send_tagstruct(s->connection->pstream, t);
 585
 586 /*             pa_log("Requesting %u bytes", l);     */
 587             break;
 588         }
 589
 590         case PLAYBACK_STREAM_MESSAGE_UNDERFLOW: {
 591             pa_tagstruct *t;
 592
 593             /* Report that we're empty */
 594             t = pa_tagstruct_new(NULL, 0);
 595             pa_tagstruct_putu32(t, PA_COMMAND_UNDERFLOW);
 596             pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 597             pa_tagstruct_putu32(t, s->index);
 598             pa_pstream_send_tagstruct(s->connection->pstream, t);
 599             break;
 600         }
 601
 602         case PLAYBACK_STREAM_MESSAGE_OVERFLOW: {
 603             pa_tagstruct *t;
 604
 605             /* Notify the user we're overflowed*/
 606             t = pa_tagstruct_new(NULL, 0);
 607             pa_tagstruct_putu32(t, PA_COMMAND_OVERFLOW);
 608             pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 609             pa_tagstruct_putu32(t, s->index);
 610             pa_pstream_send_tagstruct(s->connection->pstream, t);
 611             break;
 612         }
 613
 614         case PLAYBACK_STREAM_MESSAGE_DRAIN_ACK:
 615             pa_pstream_send_simple_ack(s->connection->pstream, PA_PTR_TO_UINT(userdata));
 616             break;
 617
 618     }
 619
 620     return 0;
 621 }
 622
 623 static playback_stream* playback_stream_new(
 624         connection *c,
 625         pa_sink *sink,
 626         pa_sample_spec *ss,
 627         pa_channel_map *map,
 628         const char *name,
 629         uint32_t *maxlength,
 630         uint32_t *tlength,
 631         uint32_t *prebuf,
 632         uint32_t *minreq,
 633         pa_cvolume *volume,
 634         uint32_t syncid,
 635         uint32_t *missing,
 636         pa_sink_input_flags_t flags) {
 637
 638     playback_stream *s, *ssync;
 639     pa_sink_input *sink_input;
 640     pa_memblock *silence;
 641     uint32_t idx;
 642     int64_t start_index;
 643     pa_sink_input_new_data data;
 644
 645     pa_assert(c);
 646     pa_assert(ss);
 647     pa_assert(name);
 648     pa_assert(maxlength);
 649
 650     /* Find syncid group */
 651     for (ssync = pa_idxset_first(c->output_streams, &idx); ssync; ssync = pa_idxset_next(c->output_streams, &idx)) {
 652
 653         if (!playback_stream_isinstance(ssync))
 654             continue;
 655
 656         if (ssync->syncid == syncid)
 657             break;
 658     }
 659
 660     /* Synced streams must connect to the same sink */
 661     if (ssync) {
 662
 663         if (!sink)
 664             sink = ssync->sink_input->sink;
 665         else if (sink != ssync->sink_input->sink)
 666             return NULL;
 667     }
 668
 669     pa_sink_input_new_data_init(&data);
 670     data.sink = sink;
 671     data.driver = __FILE__;
 672     data.name = name;
 673     pa_sink_input_new_data_set_sample_spec(&data, ss);
 674     pa_sink_input_new_data_set_channel_map(&data, map);
 675     pa_sink_input_new_data_set_volume(&data, volume);
 676     data.module = c->protocol->module;
 677     data.client = c->client;
 678     data.sync_base = ssync ? ssync->sink_input : NULL;
 679
 680     if (!(sink_input = pa_sink_input_new(c->protocol->core, &data, flags)))
 681         return NULL;
 682
 683     s = pa_msgobject_new(playback_stream);
 684     s->parent.parent.parent.free = playback_stream_free;
 685     s->parent.parent.process_msg = playback_stream_process_msg;
 686     s->connection = c;
 687     s->syncid = syncid;
 688     s->sink_input = sink_input;
 689     s->underrun = 1;
 690
 691     s->sink_input->parent.process_msg = sink_input_process_msg;
 692     s->sink_input->peek = sink_input_peek_cb;
 693     s->sink_input->drop = sink_input_drop_cb;
 694     s->sink_input->kill = sink_input_kill_cb;
 695     s->sink_input->moved = sink_input_moved_cb;
 696     s->sink_input->suspend = sink_input_suspend_cb;
 697     s->sink_input->userdata = s;
 698
 699     start_index = ssync ? pa_memblockq_get_read_index(ssync->memblockq) : 0;
 700
 701     silence = pa_silence_memblock_new(c->protocol->core->mempool, &s->sink_input->sample_spec, 0);
 702
 703     s->memblockq = pa_memblockq_new(
 704             start_index,
 705             *maxlength,
 706             *tlength,
 707             pa_frame_size(&s->sink_input->sample_spec),
 708             *prebuf,
 709             *minreq,
 710             silence);
 711
 712     pa_memblock_unref(silence);
 713
 714     *maxlength = (uint32_t) pa_memblockq_get_maxlength(s->memblockq);
 715     *tlength = (uint32_t) pa_memblockq_get_tlength(s->memblockq);
 716     *prebuf = (uint32_t) pa_memblockq_get_prebuf(s->memblockq);
 717     *minreq = (uint32_t) pa_memblockq_get_minreq(s->memblockq);
 718     *missing = (uint32_t) pa_memblockq_pop_missing(s->memblockq);
 719
 720     *ss = s->sink_input->sample_spec;
 721     *map = s->sink_input->channel_map;
 722
 723     s->minreq = pa_memblockq_get_minreq(s->memblockq);
 724     pa_atomic_store(&s->missing, 0);
 725     s->drain_request = 0;
 726
 727     pa_idxset_put(c->output_streams, s, &s->index);
 728
 729     pa_sink_input_put(s->sink_input);
 730
 731     return s;
 732 }
 733
 734 static int connection_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
 735     connection *c = CONNECTION(o);
 736     connection_assert_ref(c);
 737
 738     if (!c->protocol)
 739         return -1;
 740
 741     switch (code) {
 742
 743         case CONNECTION_MESSAGE_REVOKE:
 744             pa_pstream_send_revoke(c->pstream, PA_PTR_TO_UINT(userdata));
 745             break;
 746
 747         case CONNECTION_MESSAGE_RELEASE:
 748             pa_pstream_send_release(c->pstream, PA_PTR_TO_UINT(userdata));
 749             break;
 750     }
 751
 752     return 0;
 753 }
 754
 755 static void connection_unlink(connection *c) {
 756     record_stream *r;
 757     output_stream *o;
 758
 759     pa_assert(c);
 760
 761     if (!c->protocol)
 762         return;
 763
 764     while ((r = pa_idxset_first(c->record_streams, NULL)))
 765         record_stream_unlink(r);
 766
 767     while ((o = pa_idxset_first(c->output_streams, NULL)))
 768         if (playback_stream_isinstance(o))
 769             playback_stream_unlink(PLAYBACK_STREAM(o));
 770         else
 771             upload_stream_unlink(UPLOAD_STREAM(o));
 772
 773     if (c->subscription)
 774         pa_subscription_free(c->subscription);
 775
 776     if (c->pstream)
 777         pa_pstream_unlink(c->pstream);
 778
 779     if (c->auth_timeout_event) {
 780         c->protocol->core->mainloop->time_free(c->auth_timeout_event);
 781         c->auth_timeout_event = NULL;
 782     }
 783
 784     pa_assert_se(pa_idxset_remove_by_data(c->protocol->connections, c, NULL) == c);
 785     c->protocol = NULL;
 786     connection_unref(c);
 787 }
 788
 789 static void connection_free(pa_object *o) {
 790     connection *c = CONNECTION(o);
 791
 792     pa_assert(c);
 793
 794     connection_unlink(c);
 795
 796     pa_idxset_free(c->record_streams, NULL, NULL);
 797     pa_idxset_free(c->output_streams, NULL, NULL);
 798
 799     pa_pdispatch_unref(c->pdispatch);
 800     pa_pstream_unref(c->pstream);
 801     pa_client_free(c->client);
 802
 803     pa_xfree(c);
 804 }
 805
 806 /* Called from thread context */
 807 static void request_bytes(playback_stream *s) {
 808     size_t m, previous_missing;
 809
 810     playback_stream_assert_ref(s);
 811
 812     m = pa_memblockq_pop_missing(s->memblockq);
 813
 814     if (m <= 0)
 815         return;
 816
 817 /*     pa_log("request_bytes(%u)", m); */
 818
 819     previous_missing = pa_atomic_add(&s->missing, m);
 820     if (previous_missing < s->minreq && previous_missing+m >= s->minreq) {
 821         pa_assert(pa_thread_mq_get());
 822         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
 823     }
 824 }
 825
 826 static void send_memblock(connection *c) {
 827     uint32_t start;
 828     record_stream *r;
 829
 830     start = PA_IDXSET_INVALID;
 831     for (;;) {
 832         pa_memchunk chunk;
 833
 834         if (!(r = RECORD_STREAM(pa_idxset_rrobin(c->record_streams, &c->rrobin_index))))
 835             return;
 836
 837         if (start == PA_IDXSET_INVALID)
 838             start = c->rrobin_index;
 839         else if (start == c->rrobin_index)
 840             return;
 841
 842         if (pa_memblockq_peek(r->memblockq,  &chunk) >= 0) {
 843             pa_memchunk schunk = chunk;
 844
 845             if (schunk.length > r->fragment_size)
 846                 schunk.length = r->fragment_size;
 847
 848             pa_pstream_send_memblock(c->pstream, r->index, 0, PA_SEEK_RELATIVE, &schunk);
 849
 850             pa_memblockq_drop(r->memblockq, schunk.length);
 851             pa_memblock_unref(schunk.memblock);
 852
 853             return;
 854         }
 855     }
 856 }
 857
 858 static void send_playback_stream_killed(playback_stream *p) {
 859     pa_tagstruct *t;
 860     playback_stream_assert_ref(p);
 861
 862     t = pa_tagstruct_new(NULL, 0);
 863     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_KILLED);
 864     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 865     pa_tagstruct_putu32(t, p->index);
 866     pa_pstream_send_tagstruct(p->connection->pstream, t);
 867 }
 868
 869 static void send_record_stream_killed(record_stream *r) {
 870     pa_tagstruct *t;
 871     record_stream_assert_ref(r);
 872
 873     t = pa_tagstruct_new(NULL, 0);
 874     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_KILLED);
 875     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
 876     pa_tagstruct_putu32(t, r->index);
 877     pa_pstream_send_tagstruct(r->connection->pstream, t);
 878 }
 879
 880 /*** sink input callbacks ***/
 881
 882 /* Called from thread context */
 883 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
 884     pa_sink_input *i = PA_SINK_INPUT(o);
 885     playback_stream *s;
 886
 887     pa_sink_input_assert_ref(i);
 888     s = PLAYBACK_STREAM(i->userdata);
 889     playback_stream_assert_ref(s);
 890
 891     switch (code) {
 892
 893         case SINK_INPUT_MESSAGE_SEEK:
 894             pa_memblockq_seek(s->memblockq, offset, PA_PTR_TO_UINT(userdata));
 895             request_bytes(s);
 896             return 0;
 897
 898         case SINK_INPUT_MESSAGE_POST_DATA: {
 899             pa_assert(chunk);
 900
 901 /*             pa_log("sink input post: %u", chunk->length); */
 902
 903             if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
 904
 905                 pa_log_warn("Failed to push data into queue");
 906                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_OVERFLOW, NULL, 0, NULL, NULL);
 907                 pa_memblockq_seek(s->memblockq, chunk->length, PA_SEEK_RELATIVE);
 908             }
 909
 910             request_bytes(s);
 911
 912             s->underrun = 0;
 913             return 0;
 914         }
 915
 916         case SINK_INPUT_MESSAGE_DRAIN: {
 917
 918             pa_memblockq_prebuf_disable(s->memblockq);
 919
 920             if (!pa_memblockq_is_readable(s->memblockq))
 921                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_DRAIN_ACK, userdata, 0, NULL, NULL);
 922             else {
 923                 s->drain_tag = PA_PTR_TO_UINT(userdata);
 924                 s->drain_request = 1;
 925             }
 926             request_bytes(s);
 927
 928             return 0;
 929         }
 930
 931         case SINK_INPUT_MESSAGE_FLUSH:
 932         case SINK_INPUT_MESSAGE_PREBUF_FORCE:
 933         case SINK_INPUT_MESSAGE_TRIGGER: {
 934
 935             pa_sink_input *isync;
 936             void (*func)(pa_memblockq *bq);
 937
 938             switch  (code) {
 939                 case SINK_INPUT_MESSAGE_FLUSH:
 940                     func = pa_memblockq_flush;
 941                     break;
 942
 943                 case SINK_INPUT_MESSAGE_PREBUF_FORCE:
 944                     func = pa_memblockq_prebuf_force;
 945                     break;
 946
 947                 case SINK_INPUT_MESSAGE_TRIGGER:
 948                     func = pa_memblockq_prebuf_disable;
 949                     break;
 950
 951                 default:
 952                     pa_assert_not_reached();
 953             }
 954
 955             func(s->memblockq);
 956             s->underrun = 0;
 957             request_bytes(s);
 958
 959             /* Do the same for all other members in the sync group */
 960             for (isync = i->sync_prev; isync; isync = isync->sync_prev) {
 961                 playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
 962                 func(ssync->memblockq);
 963                 ssync->underrun = 0;
 964                 request_bytes(ssync);
 965             }
 966
 967             for (isync = i->sync_next; isync; isync = isync->sync_next) {
 968                 playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
 969                 func(ssync->memblockq);
 970                 ssync->underrun = 0;
 971                 request_bytes(ssync);
 972             }
 973
 974             return 0;
 975         }
 976
 977         case SINK_INPUT_MESSAGE_UPDATE_LATENCY:
 978
 979             s->read_index = pa_memblockq_get_read_index(s->memblockq);
 980             s->write_index = pa_memblockq_get_write_index(s->memblockq);
 981             s->resampled_chunk_length = s->sink_input->thread_info.resampled_chunk.memblock ? s->sink_input->thread_info.resampled_chunk.length : 0;
 982             return 0;
 983
 984         case PA_SINK_INPUT_MESSAGE_SET_STATE:
 985
 986             pa_memblockq_prebuf_force(s->memblockq);
 987             request_bytes(s);
 988             break;
 989
 990         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
 991             pa_usec_t *r = userdata;
 992
 993             *r = pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
 994
 995             /* Fall through, the default handler will add in the extra
 996              * latency added by the resampler */
 997             break;
 998         }
 999     }
1000
1001     return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
1002 }
1003
1004 /* Called from thread context */
1005 static int sink_input_peek_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
1006     playback_stream *s;
1007
1008     pa_sink_input_assert_ref(i);
1009     s = PLAYBACK_STREAM(i->userdata);
1010     playback_stream_assert_ref(s);
1011     pa_assert(chunk);
1012
1013     if (pa_memblockq_get_length(s->memblockq) <= 0 && !s->underrun) {
1014         pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_UNDERFLOW, NULL, 0, NULL, NULL);
1015         s->underrun = 1;
1016     }
1017
1018     if (pa_memblockq_peek(s->memblockq, chunk) < 0) {
1019 /*         pa_log("peek: failure");     */
1020         return -1;
1021     }
1022
1023 /*     pa_log("peek: %u", chunk->length); */
1024
1025     request_bytes(s);
1026
1027     return 0;
1028 }
1029
1030 /* Called from thread context */
1031 static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
1032     playback_stream *s;
1033
1034     pa_sink_input_assert_ref(i);
1035     s = PLAYBACK_STREAM(i->userdata);
1036     playback_stream_assert_ref(s);
1037     pa_assert(length > 0);
1038
1039     pa_memblockq_drop(s->memblockq, length);
1040
1041     if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) {
1042         s->drain_request = 0;
1043         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);
1044     }
1045
1046     request_bytes(s);
1047
1048 /*     pa_log("after_drop: %u %u", pa_memblockq_get_length(s->memblockq), pa_memblockq_is_readable(s->memblockq)); */
1049 }
1050
1051 /* Called from main context */
1052 static void sink_input_kill_cb(pa_sink_input *i) {
1053     playback_stream *s;
1054
1055     pa_sink_input_assert_ref(i);
1056     s = PLAYBACK_STREAM(i->userdata);
1057     playback_stream_assert_ref(s);
1058
1059     send_playback_stream_killed(s);
1060     playback_stream_unlink(s);
1061 }
1062
1063 /* Called from main context */
1064 static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspend) {
1065     playback_stream *s;
1066     pa_tagstruct *t;
1067
1068     pa_sink_input_assert_ref(i);
1069     s = PLAYBACK_STREAM(i->userdata);
1070     playback_stream_assert_ref(s);
1071
1072     t = pa_tagstruct_new(NULL, 0);
1073     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_SUSPENDED);
1074     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1075     pa_tagstruct_putu32(t, s->index);
1076     pa_tagstruct_put_boolean(t, suspend);
1077     pa_pstream_send_tagstruct(s->connection->pstream, t);
1078 }
1079
1080 /* Called from main context */
1081 static void sink_input_moved_cb(pa_sink_input *i) {
1082     playback_stream *s;
1083     pa_tagstruct *t;
1084
1085     pa_sink_input_assert_ref(i);
1086     s = PLAYBACK_STREAM(i->userdata);
1087     playback_stream_assert_ref(s);
1088
1089     t = pa_tagstruct_new(NULL, 0);
1090     pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_MOVED);
1091     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1092     pa_tagstruct_putu32(t, s->index);
1093     pa_tagstruct_putu32(t, i->sink->index);
1094     pa_tagstruct_puts(t, i->sink->name);
1095     pa_tagstruct_put_boolean(t, pa_sink_get_state(i->sink) == PA_SINK_SUSPENDED);
1096     pa_pstream_send_tagstruct(s->connection->pstream, t);
1097 }
1098
1099 /*** source_output callbacks ***/
1100
1101 /* Called from thread context */
1102 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
1103     record_stream *s;
1104
1105     pa_source_output_assert_ref(o);
1106     s = RECORD_STREAM(o->userdata);
1107     record_stream_assert_ref(s);
1108     pa_assert(chunk);
1109
1110     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), RECORD_STREAM_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
1111 }
1112
1113 static void source_output_kill_cb(pa_source_output *o) {
1114     record_stream *s;
1115
1116     pa_source_output_assert_ref(o);
1117     s = RECORD_STREAM(o->userdata);
1118     record_stream_assert_ref(s);
1119
1120     send_record_stream_killed(s);
1121     record_stream_unlink(s);
1122 }
1123
1124 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
1125     record_stream *s;
1126
1127     pa_source_output_assert_ref(o);
1128     s = RECORD_STREAM(o->userdata);
1129     record_stream_assert_ref(s);
1130
1131     /*pa_log("get_latency: %u", pa_memblockq_get_length(s->memblockq));*/
1132
1133     return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &o->sample_spec);
1134 }
1135
1136 /* Called from main context */
1137 static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend) {
1138     record_stream *s;
1139     pa_tagstruct *t;
1140
1141     pa_source_output_assert_ref(o);
1142     s = RECORD_STREAM(o->userdata);
1143     record_stream_assert_ref(s);
1144
1145     t = pa_tagstruct_new(NULL, 0);
1146     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_SUSPENDED);
1147     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1148     pa_tagstruct_putu32(t, s->index);
1149     pa_tagstruct_put_boolean(t, suspend);
1150     pa_pstream_send_tagstruct(s->connection->pstream, t);
1151 }
1152
1153 /* Called from main context */
1154 static void source_output_moved_cb(pa_source_output *o) {
1155     record_stream *s;
1156     pa_tagstruct *t;
1157
1158     pa_source_output_assert_ref(o);
1159     s = RECORD_STREAM(o->userdata);
1160     record_stream_assert_ref(s);
1161
1162     t = pa_tagstruct_new(NULL, 0);
1163     pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_MOVED);
1164     pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
1165     pa_tagstruct_putu32(t, s->index);
1166     pa_tagstruct_putu32(t, o->source->index);
1167     pa_tagstruct_puts(t, o->source->name);
1168     pa_pstream_send_tagstruct(s->connection->pstream, t);
1169 }
1170
1171 /*** pdispatch callbacks ***/
1172
1173 static void protocol_error(connection *c) {
1174     pa_log("protocol error, kicking client");
1175     connection_unlink(c);
1176 }
1177
1178 #define CHECK_VALIDITY(pstream, expression, tag, error) do { \
1179 if (!(expression)) { \
1180     pa_pstream_send_error((pstream), (tag), (error)); \
1181     return; \
1182 } \
1183 } while(0);
1184
1185 static pa_tagstruct *reply_new(uint32_t tag) {
1186     pa_tagstruct *reply;
1187
1188     reply = pa_tagstruct_new(NULL, 0);
1189     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1190     pa_tagstruct_putu32(reply, tag);
1191     return reply;
1192 }
1193
1194 static void command_create_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1195     connection *c = CONNECTION(userdata);
1196     playback_stream *s;
1197     uint32_t maxlength, tlength, prebuf, minreq, sink_index, syncid, missing;
1198     const char *name, *sink_name;
1199     pa_sample_spec ss;
1200     pa_channel_map map;
1201     pa_tagstruct *reply;
1202     pa_sink *sink = NULL;
1203     pa_cvolume volume;
1204     int corked;
1205     int no_remap = 0, no_remix = 0, fix_format = 0, fix_rate = 0, fix_channels = 0, no_move = 0, variable_rate = 0;
1206     pa_sink_input_flags_t flags = 0;
1207
1208     connection_assert_ref(c);
1209     pa_assert(t);
1210
1211     if (pa_tagstruct_get(
1212             t,
1213             PA_TAG_STRING, &name,
1214             PA_TAG_SAMPLE_SPEC, &ss,
1215             PA_TAG_CHANNEL_MAP, &map,
1216             PA_TAG_U32, &sink_index,
1217             PA_TAG_STRING, &sink_name,
1218             PA_TAG_U32, &maxlength,
1219             PA_TAG_BOOLEAN, &corked,
1220             PA_TAG_U32, &tlength,
1221             PA_TAG_U32, &prebuf,
1222             PA_TAG_U32, &minreq,
1223             PA_TAG_U32, &syncid,
1224             PA_TAG_CVOLUME, &volume,
1225             PA_TAG_INVALID) < 0 || !name) {
1226         protocol_error(c);
1227         return;
1228     }
1229
1230     if (c->version >= 12)  {
1231         /* Since 0.9.8 the user can ask for a couple of additional flags */
1232
1233         if (pa_tagstruct_get_boolean(t, &no_remap) < 0 ||
1234             pa_tagstruct_get_boolean(t, &no_remix) < 0 ||
1235             pa_tagstruct_get_boolean(t, &fix_format) < 0 ||
1236             pa_tagstruct_get_boolean(t, &fix_rate) < 0 ||
1237             pa_tagstruct_get_boolean(t, &fix_channels) < 0 ||
1238             pa_tagstruct_get_boolean(t, &no_move) < 0 ||
1239             pa_tagstruct_get_boolean(t, &variable_rate) < 0) {
1240             protocol_error(c);
1241             return;
1242         }
1243     }
1244
1245     if (!pa_tagstruct_eof(t)) {
1246         protocol_error(c);
1247         return;
1248     }
1249
1250     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1251     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1252     CHECK_VALIDITY(c->pstream, sink_index != PA_INVALID_INDEX || !sink_name || (*sink_name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
1253     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1254     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1255     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
1256     CHECK_VALIDITY(c->pstream, map.channels == ss.channels && volume.channels == ss.channels, tag, PA_ERR_INVALID);
1257     CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
1258     CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
1259
1260     if (sink_index != PA_INVALID_INDEX) {
1261         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
1262         CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
1263     } else if (sink_name) {
1264         sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1);
1265         CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
1266     }
1267
1268     flags =
1269         (corked ?  PA_SINK_INPUT_START_CORKED : 0) |
1270         (no_remap ?  PA_SINK_INPUT_NO_REMAP : 0) |
1271         (no_remix ?  PA_SINK_INPUT_NO_REMIX : 0) |
1272         (fix_format ?  PA_SINK_INPUT_FIX_FORMAT : 0) |
1273         (fix_rate ?  PA_SINK_INPUT_FIX_RATE : 0) |
1274         (fix_channels ?  PA_SINK_INPUT_FIX_CHANNELS : 0) |
1275         (no_move ?  PA_SINK_INPUT_DONT_MOVE : 0) |
1276         (variable_rate ?  PA_SINK_INPUT_VARIABLE_RATE : 0);
1277
1278     s = playback_stream_new(c, sink, &ss, &map, name, &maxlength, &tlength, &prebuf, &minreq, &volume, syncid, &missing, flags);
1279     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
1280
1281     reply = reply_new(tag);
1282     pa_tagstruct_putu32(reply, s->index);
1283     pa_assert(s->sink_input);
1284     pa_tagstruct_putu32(reply, s->sink_input->index);
1285     pa_tagstruct_putu32(reply, missing);
1286
1287 /*     pa_log("initial request is %u", missing); */
1288
1289     if (c->version >= 9) {
1290         /* Since 0.9.0 we support sending the buffer metrics back to the client */
1291
1292         pa_tagstruct_putu32(reply, (uint32_t) maxlength);
1293         pa_tagstruct_putu32(reply, (uint32_t) tlength);
1294         pa_tagstruct_putu32(reply, (uint32_t) prebuf);
1295         pa_tagstruct_putu32(reply, (uint32_t) minreq);
1296     }
1297
1298     if (c->version >= 12) {
1299         /* Since 0.9.8 we support sending the chosen sample
1300          * spec/channel map/device/suspend status back to the
1301          * client */
1302
1303         pa_tagstruct_put_sample_spec(reply, &ss);
1304         pa_tagstruct_put_channel_map(reply, &map);
1305
1306         pa_tagstruct_putu32(reply, s->sink_input->sink->index);
1307         pa_tagstruct_puts(reply, s->sink_input->sink->name);
1308
1309         pa_tagstruct_put_boolean(reply, pa_sink_get_state(s->sink_input->sink) == PA_SINK_SUSPENDED);
1310     }
1311
1312     pa_pstream_send_tagstruct(c->pstream, reply);
1313 }
1314
1315 static void command_delete_stream(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1316     connection *c = CONNECTION(userdata);
1317     uint32_t channel;
1318
1319     connection_assert_ref(c);
1320     pa_assert(t);
1321
1322     if (pa_tagstruct_getu32(t, &channel) < 0 ||
1323         !pa_tagstruct_eof(t)) {
1324         protocol_error(c);
1325         return;
1326     }
1327
1328     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1329
1330     switch (command) {
1331
1332         case PA_COMMAND_DELETE_PLAYBACK_STREAM: {
1333             playback_stream *s;
1334             if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !playback_stream_isinstance(s)) {
1335                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
1336                 return;
1337             }
1338
1339             playback_stream_unlink(s);
1340             break;
1341         }
1342
1343         case PA_COMMAND_DELETE_RECORD_STREAM: {
1344             record_stream *s;
1345             if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
1346                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
1347                 return;
1348             }
1349
1350             record_stream_unlink(s);
1351             break;
1352         }
1353
1354         case PA_COMMAND_DELETE_UPLOAD_STREAM: {
1355             upload_stream *s;
1356
1357             if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !upload_stream_isinstance(s)) {
1358                 pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
1359                 return;
1360             }
1361
1362             upload_stream_unlink(s);
1363             break;
1364         }
1365
1366         default:
1367             pa_assert_not_reached();
1368     }
1369
1370     pa_pstream_send_simple_ack(c->pstream, tag);
1371 }
1372
1373 static void command_create_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1374     connection *c = CONNECTION(userdata);
1375     record_stream *s;
1376     uint32_t maxlength, fragment_size;
1377     uint32_t source_index;
1378     const char *name, *source_name;
1379     pa_sample_spec ss;
1380     pa_channel_map map;
1381     pa_tagstruct *reply;
1382     pa_source *source = NULL;
1383     int corked;
1384     int no_remap = 0, no_remix = 0, fix_format = 0, fix_rate = 0, fix_channels = 0, no_move = 0, variable_rate = 0;
1385     pa_source_output_flags_t flags = 0;
1386
1387     connection_assert_ref(c);
1388     pa_assert(t);
1389
1390     if (pa_tagstruct_gets(t, &name) < 0 ||
1391         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1392         pa_tagstruct_get_channel_map(t, &map) < 0 ||
1393         pa_tagstruct_getu32(t, &source_index) < 0 ||
1394         pa_tagstruct_gets(t, &source_name) < 0 ||
1395         pa_tagstruct_getu32(t, &maxlength) < 0 ||
1396         pa_tagstruct_get_boolean(t, &corked) < 0 ||
1397         pa_tagstruct_getu32(t, &fragment_size) < 0) {
1398         protocol_error(c);
1399         return;
1400     }
1401
1402     if (c->version >= 12)  {
1403         /* Since 0.9.8 the user can ask for a couple of additional flags */
1404
1405         if (pa_tagstruct_get_boolean(t, &no_remap) < 0 ||
1406             pa_tagstruct_get_boolean(t, &no_remix) < 0 ||
1407             pa_tagstruct_get_boolean(t, &fix_format) < 0 ||
1408             pa_tagstruct_get_boolean(t, &fix_rate) < 0 ||
1409             pa_tagstruct_get_boolean(t, &fix_channels) < 0 ||
1410             pa_tagstruct_get_boolean(t, &no_move) < 0 ||
1411             pa_tagstruct_get_boolean(t, &variable_rate) < 0) {
1412             protocol_error(c);
1413             return;
1414         }
1415     }
1416
1417     if (!pa_tagstruct_eof(t)) {
1418         protocol_error(c);
1419         return;
1420     }
1421
1422     flags =
1423         (corked ?  PA_SOURCE_OUTPUT_START_CORKED : 0) |
1424         (no_remap ?  PA_SOURCE_OUTPUT_NO_REMAP : 0) |
1425         (no_remix ?  PA_SOURCE_OUTPUT_NO_REMIX : 0) |
1426         (fix_format ?  PA_SOURCE_OUTPUT_FIX_FORMAT : 0) |
1427         (fix_rate ?  PA_SOURCE_OUTPUT_FIX_RATE : 0) |
1428         (fix_channels ?  PA_SOURCE_OUTPUT_FIX_CHANNELS : 0) |
1429         (no_move ?  PA_SOURCE_OUTPUT_DONT_MOVE : 0) |
1430         (variable_rate ?  PA_SOURCE_OUTPUT_VARIABLE_RATE : 0);
1431
1432     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1433     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1434     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1435     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1436     CHECK_VALIDITY(c->pstream, source_index != PA_INVALID_INDEX || !source_name || (*source_name && pa_utf8_valid(source_name)), tag, PA_ERR_INVALID);
1437     CHECK_VALIDITY(c->pstream, map.channels == ss.channels, tag, PA_ERR_INVALID);
1438     CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
1439     CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
1440
1441     if (source_index != PA_INVALID_INDEX) {
1442         source = pa_idxset_get_by_index(c->protocol->core->sources, source_index);
1443         CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
1444     } else if (source_name) {
1445         source = pa_namereg_get(c->protocol->core, source_name, PA_NAMEREG_SOURCE, 1);
1446         CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
1447     }
1448
1449     s = record_stream_new(c, source, &ss, &map, name, &maxlength, fragment_size, flags);
1450     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
1451
1452     reply = reply_new(tag);
1453     pa_tagstruct_putu32(reply, s->index);
1454     pa_assert(s->source_output);
1455     pa_tagstruct_putu32(reply, s->source_output->index);
1456
1457     if (c->version >= 9) {
1458         /* Since 0.9 we support sending the buffer metrics back to the client */
1459
1460         pa_tagstruct_putu32(reply, (uint32_t) maxlength);
1461         pa_tagstruct_putu32(reply, (uint32_t) s->fragment_size);
1462     }
1463
1464     if (c->version >= 12) {
1465         /* Since 0.9.8 we support sending the chosen sample
1466          * spec/channel map/device/suspend status back to the
1467          * client */
1468
1469         pa_tagstruct_put_sample_spec(reply, &ss);
1470         pa_tagstruct_put_channel_map(reply, &map);
1471
1472         pa_tagstruct_putu32(reply, s->source_output->source->index);
1473         pa_tagstruct_puts(reply, s->source_output->source->name);
1474
1475         pa_tagstruct_put_boolean(reply, pa_source_get_state(s->source_output->source) == PA_SOURCE_SUSPENDED);
1476     }
1477
1478     pa_pstream_send_tagstruct(c->pstream, reply);
1479 }
1480
1481 static void command_exit(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1482     connection *c = CONNECTION(userdata);
1483
1484     connection_assert_ref(c);
1485     pa_assert(t);
1486
1487     if (!pa_tagstruct_eof(t)) {
1488         protocol_error(c);
1489         return;
1490     }
1491
1492     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1493
1494     c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
1495     pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
1496 }
1497
1498 static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1499     connection *c = CONNECTION(userdata);
1500     const void*cookie;
1501     pa_tagstruct *reply;
1502
1503     connection_assert_ref(c);
1504     pa_assert(t);
1505
1506     if (pa_tagstruct_getu32(t, &c->version) < 0 ||
1507         pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
1508         !pa_tagstruct_eof(t)) {
1509         protocol_error(c);
1510         return;
1511     }
1512
1513     /* Minimum supported version */
1514     if (c->version < 8) {
1515         pa_pstream_send_error(c->pstream, tag, PA_ERR_VERSION);
1516         return;
1517     }
1518
1519     if (!c->authorized) {
1520         int success = 0;
1521
1522 #ifdef HAVE_CREDS
1523         const pa_creds *creds;
1524
1525         if ((creds = pa_pdispatch_creds(pd))) {
1526             if (creds->uid == getuid())
1527                 success = 1;
1528             else if (c->protocol->auth_group) {
1529                 int r;
1530                 gid_t gid;
1531
1532                 if ((gid = pa_get_gid_of_group(c->protocol->auth_group)) == (gid_t) -1)
1533                     pa_log_warn("failed to get GID of group '%s'", c->protocol->auth_group);
1534                 else if (gid == creds->gid)
1535                     success = 1;
1536
1537                 if (!success) {
1538                     if ((r = pa_uid_in_group(creds->uid, c->protocol->auth_group)) < 0)
1539                         pa_log_warn("failed to check group membership.");
1540                     else if (r > 0)
1541                         success = 1;
1542                 }
1543             }
1544
1545             pa_log_info("Got credentials: uid=%lu gid=%lu success=%i",
1546                         (unsigned long) creds->uid,
1547                         (unsigned long) creds->gid,
1548                         success);
1549
1550             if (c->version >= 10 &&
1551                 pa_mempool_is_shared(c->protocol->core->mempool) &&
1552                 creds->uid == getuid()) {
1553
1554                 pa_pstream_use_shm(c->pstream, 1);
1555                 pa_log_info("Enabled SHM for new connection");
1556             }
1557
1558         }
1559 #endif
1560
1561         if (!success && memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) == 0)
1562             success = 1;
1563
1564         if (!success) {
1565             pa_log_warn("Denied access to client with invalid authorization data.");
1566             pa_pstream_send_error(c->pstream, tag, PA_ERR_ACCESS);
1567             return;
1568         }
1569
1570         c->authorized = 1;
1571         if (c->auth_timeout_event) {
1572             c->protocol->core->mainloop->time_free(c->auth_timeout_event);
1573             c->auth_timeout_event = NULL;
1574         }
1575     }
1576
1577     reply = reply_new(tag);
1578     pa_tagstruct_putu32(reply, PA_PROTOCOL_VERSION);
1579
1580 #ifdef HAVE_CREDS
1581 {
1582     /* SHM support is only enabled after both sides made sure they are the same user. */
1583
1584     pa_creds ucred;
1585
1586     ucred.uid = getuid();
1587     ucred.gid = getgid();
1588
1589     pa_pstream_send_tagstruct_with_creds(c->pstream, reply, &ucred);
1590 }
1591 #else
1592     pa_pstream_send_tagstruct(c->pstream, reply);
1593 #endif
1594 }
1595
1596 static void command_set_client_name(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1597     connection *c = CONNECTION(userdata);
1598     const char *name;
1599
1600     connection_assert_ref(c);
1601     pa_assert(t);
1602
1603     if (pa_tagstruct_gets(t, &name) < 0 ||
1604         !pa_tagstruct_eof(t)) {
1605         protocol_error(c);
1606         return;
1607     }
1608
1609     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1610
1611     pa_client_set_name(c->client, name);
1612     pa_pstream_send_simple_ack(c->pstream, tag);
1613 }
1614
1615 static void command_lookup(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1616     connection *c = CONNECTION(userdata);
1617     const char *name;
1618     uint32_t idx = PA_IDXSET_INVALID;
1619
1620     connection_assert_ref(c);
1621     pa_assert(t);
1622
1623     if (pa_tagstruct_gets(t, &name) < 0 ||
1624         !pa_tagstruct_eof(t)) {
1625         protocol_error(c);
1626         return;
1627     }
1628
1629     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1630     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1631
1632     if (command == PA_COMMAND_LOOKUP_SINK) {
1633         pa_sink *sink;
1634         if ((sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1)))
1635             idx = sink->index;
1636     } else {
1637         pa_source *source;
1638         pa_assert(command == PA_COMMAND_LOOKUP_SOURCE);
1639         if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1)))
1640             idx = source->index;
1641     }
1642
1643     if (idx == PA_IDXSET_INVALID)
1644         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1645     else {
1646         pa_tagstruct *reply;
1647         reply = reply_new(tag);
1648         pa_tagstruct_putu32(reply, idx);
1649         pa_pstream_send_tagstruct(c->pstream, reply);
1650     }
1651 }
1652
1653 static void command_drain_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1654     connection *c = CONNECTION(userdata);
1655     uint32_t idx;
1656     playback_stream *s;
1657
1658     connection_assert_ref(c);
1659     pa_assert(t);
1660
1661     if (pa_tagstruct_getu32(t, &idx) < 0 ||
1662         !pa_tagstruct_eof(t)) {
1663         protocol_error(c);
1664         return;
1665     }
1666
1667     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1668     s = pa_idxset_get_by_index(c->output_streams, idx);
1669     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1670     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
1671
1672     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);
1673 }
1674
1675 static void command_stat(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1676     connection *c = CONNECTION(userdata);
1677     pa_tagstruct *reply;
1678     const pa_mempool_stat *stat;
1679
1680     connection_assert_ref(c);
1681     pa_assert(t);
1682
1683     if (!pa_tagstruct_eof(t)) {
1684         protocol_error(c);
1685         return;
1686     }
1687
1688     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1689
1690     stat = pa_mempool_get_stat(c->protocol->core->mempool);
1691
1692     reply = reply_new(tag);
1693     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->n_allocated));
1694     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->allocated_size));
1695     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->n_accumulated));
1696     pa_tagstruct_putu32(reply, (uint32_t) pa_atomic_load(&stat->accumulated_size));
1697     pa_tagstruct_putu32(reply, pa_scache_total_size(c->protocol->core));
1698     pa_pstream_send_tagstruct(c->pstream, reply);
1699 }
1700
1701 static void command_get_playback_latency(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1702     connection *c = CONNECTION(userdata);
1703     pa_tagstruct *reply;
1704     playback_stream *s;
1705     struct timeval tv, now;
1706     uint32_t idx;
1707     pa_usec_t latency;
1708
1709     connection_assert_ref(c);
1710     pa_assert(t);
1711
1712     if (pa_tagstruct_getu32(t, &idx) < 0 ||
1713         pa_tagstruct_get_timeval(t, &tv) < 0 ||
1714         !pa_tagstruct_eof(t)) {
1715         protocol_error(c);
1716         return;
1717     }
1718
1719     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1720     s = pa_idxset_get_by_index(c->output_streams, idx);
1721     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1722     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
1723     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)
1724
1725     reply = reply_new(tag);
1726
1727     latency = pa_sink_get_latency(s->sink_input->sink);
1728     latency += pa_bytes_to_usec(s->resampled_chunk_length, &s->sink_input->sample_spec);
1729
1730     pa_tagstruct_put_usec(reply, latency);
1731
1732     pa_tagstruct_put_usec(reply, 0);
1733     pa_tagstruct_put_boolean(reply, pa_sink_input_get_state(s->sink_input) == PA_SINK_INPUT_RUNNING);
1734     pa_tagstruct_put_timeval(reply, &tv);
1735     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
1736     pa_tagstruct_puts64(reply, s->write_index);
1737     pa_tagstruct_puts64(reply, s->read_index);
1738     pa_pstream_send_tagstruct(c->pstream, reply);
1739 }
1740
1741 static void command_get_record_latency(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1742     connection *c = CONNECTION(userdata);
1743     pa_tagstruct *reply;
1744     record_stream *s;
1745     struct timeval tv, now;
1746     uint32_t idx;
1747
1748     connection_assert_ref(c);
1749     pa_assert(t);
1750
1751     if (pa_tagstruct_getu32(t, &idx) < 0 ||
1752         pa_tagstruct_get_timeval(t, &tv) < 0 ||
1753         !pa_tagstruct_eof(t)) {
1754         protocol_error(c);
1755         return;
1756     }
1757
1758     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1759     s = pa_idxset_get_by_index(c->record_streams, idx);
1760     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1761
1762     reply = reply_new(tag);
1763     pa_tagstruct_put_usec(reply, s->source_output->source->monitor_of ? pa_sink_get_latency(s->source_output->source->monitor_of) : 0);
1764     pa_tagstruct_put_usec(reply, pa_source_get_latency(s->source_output->source));
1765     pa_tagstruct_put_boolean(reply, 0);
1766     pa_tagstruct_put_timeval(reply, &tv);
1767     pa_tagstruct_put_timeval(reply, pa_gettimeofday(&now));
1768     pa_tagstruct_puts64(reply, pa_memblockq_get_write_index(s->memblockq));
1769     pa_tagstruct_puts64(reply, pa_memblockq_get_read_index(s->memblockq));
1770     pa_pstream_send_tagstruct(c->pstream, reply);
1771 }
1772
1773 static void command_create_upload_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1774     connection *c = CONNECTION(userdata);
1775     upload_stream *s;
1776     uint32_t length;
1777     const char *name;
1778     pa_sample_spec ss;
1779     pa_channel_map map;
1780     pa_tagstruct *reply;
1781
1782     connection_assert_ref(c);
1783     pa_assert(t);
1784
1785     if (pa_tagstruct_gets(t, &name) < 0 ||
1786         pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1787         pa_tagstruct_get_channel_map(t, &map) < 0 ||
1788         pa_tagstruct_getu32(t, &length) < 0 ||
1789         !pa_tagstruct_eof(t)) {
1790         protocol_error(c);
1791         return;
1792     }
1793
1794     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1795     CHECK_VALIDITY(c->pstream, pa_sample_spec_valid(&ss), tag, PA_ERR_INVALID);
1796     CHECK_VALIDITY(c->pstream, pa_channel_map_valid(&map), tag, PA_ERR_INVALID);
1797     CHECK_VALIDITY(c->pstream, map.channels == ss.channels, tag, PA_ERR_INVALID);
1798     CHECK_VALIDITY(c->pstream, (length % pa_frame_size(&ss)) == 0 && length > 0, tag, PA_ERR_INVALID);
1799     CHECK_VALIDITY(c->pstream, length <= PA_SCACHE_ENTRY_SIZE_MAX, tag, PA_ERR_TOOLARGE);
1800     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1801
1802     s = upload_stream_new(c, &ss, &map, name, length);
1803     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
1804
1805     reply = reply_new(tag);
1806     pa_tagstruct_putu32(reply, s->index);
1807     pa_tagstruct_putu32(reply, length);
1808     pa_pstream_send_tagstruct(c->pstream, reply);
1809 }
1810
1811 static void command_finish_upload_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1812     connection *c = CONNECTION(userdata);
1813     uint32_t channel;
1814     upload_stream *s;
1815     uint32_t idx;
1816
1817     connection_assert_ref(c);
1818     pa_assert(t);
1819
1820     if (pa_tagstruct_getu32(t, &channel) < 0 ||
1821         !pa_tagstruct_eof(t)) {
1822         protocol_error(c);
1823         return;
1824     }
1825
1826     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1827
1828     s = pa_idxset_get_by_index(c->output_streams, channel);
1829     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
1830     CHECK_VALIDITY(c->pstream, upload_stream_isinstance(s), tag, PA_ERR_NOENTITY);
1831
1832     if (pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->channel_map, &s->memchunk, &idx) < 0)
1833         pa_pstream_send_error(c->pstream, tag, PA_ERR_INTERNAL);
1834     else
1835         pa_pstream_send_simple_ack(c->pstream, tag);
1836
1837     upload_stream_unlink(s);
1838 }
1839
1840 static void command_play_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1841     connection *c = CONNECTION(userdata);
1842     uint32_t sink_index;
1843     pa_volume_t volume;
1844     pa_sink *sink;
1845     const char *name, *sink_name;
1846
1847     connection_assert_ref(c);
1848     pa_assert(t);
1849
1850     if (pa_tagstruct_getu32(t, &sink_index) < 0 ||
1851         pa_tagstruct_gets(t, &sink_name) < 0 ||
1852         pa_tagstruct_getu32(t, &volume) < 0 ||
1853         pa_tagstruct_gets(t, &name) < 0 ||
1854         !pa_tagstruct_eof(t)) {
1855         protocol_error(c);
1856         return;
1857     }
1858
1859     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1860     CHECK_VALIDITY(c->pstream, sink_index != PA_INVALID_INDEX || !sink_name || (*sink_name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
1861     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1862
1863     if (sink_index != PA_INVALID_INDEX)
1864         sink = pa_idxset_get_by_index(c->protocol->core->sinks, sink_index);
1865     else
1866         sink = pa_namereg_get(c->protocol->core, sink_name, PA_NAMEREG_SINK, 1);
1867
1868     CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
1869
1870     if (pa_scache_play_item(c->protocol->core, name, sink, volume) < 0) {
1871         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1872         return;
1873     }
1874
1875     pa_pstream_send_simple_ack(c->pstream, tag);
1876 }
1877
1878 static void command_remove_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1879     connection *c = CONNECTION(userdata);
1880     const char *name;
1881
1882     connection_assert_ref(c);
1883     pa_assert(t);
1884
1885     if (pa_tagstruct_gets(t, &name) < 0 ||
1886         !pa_tagstruct_eof(t)) {
1887         protocol_error(c);
1888         return;
1889     }
1890
1891     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
1892     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
1893
1894     if (pa_scache_remove_item(c->protocol->core, name) < 0) {
1895         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
1896         return;
1897     }
1898
1899     pa_pstream_send_simple_ack(c->pstream, tag);
1900 }
1901
1902 static void sink_fill_tagstruct(pa_tagstruct *t, pa_sink *sink) {
1903     pa_assert(t);
1904     pa_sink_assert_ref(sink);
1905
1906     pa_tagstruct_put(
1907         t,
1908         PA_TAG_U32, sink->index,
1909         PA_TAG_STRING, sink->name,
1910         PA_TAG_STRING, sink->description,
1911         PA_TAG_SAMPLE_SPEC, &sink->sample_spec,
1912         PA_TAG_CHANNEL_MAP, &sink->channel_map,
1913         PA_TAG_U32, sink->module ? sink->module->index : PA_INVALID_INDEX,
1914         PA_TAG_CVOLUME, pa_sink_get_volume(sink),
1915         PA_TAG_BOOLEAN, pa_sink_get_mute(sink),
1916         PA_TAG_U32, sink->monitor_source ? sink->monitor_source->index : PA_INVALID_INDEX,
1917         PA_TAG_STRING, sink->monitor_source ? sink->monitor_source->name : NULL,
1918         PA_TAG_USEC, pa_sink_get_latency(sink),
1919         PA_TAG_STRING, sink->driver,
1920         PA_TAG_U32, sink->flags,
1921         PA_TAG_INVALID);
1922 }
1923
1924 static void source_fill_tagstruct(pa_tagstruct *t, pa_source *source) {
1925     pa_assert(t);
1926     pa_source_assert_ref(source);
1927
1928     pa_tagstruct_put(
1929         t,
1930         PA_TAG_U32, source->index,
1931         PA_TAG_STRING, source->name,
1932         PA_TAG_STRING, source->description,
1933         PA_TAG_SAMPLE_SPEC, &source->sample_spec,
1934         PA_TAG_CHANNEL_MAP, &source->channel_map,
1935         PA_TAG_U32, source->module ? source->module->index : PA_INVALID_INDEX,
1936         PA_TAG_CVOLUME, pa_source_get_volume(source),
1937         PA_TAG_BOOLEAN, pa_source_get_mute(source),
1938         PA_TAG_U32, source->monitor_of ? source->monitor_of->index : PA_INVALID_INDEX,
1939         PA_TAG_STRING, source->monitor_of ? source->monitor_of->name : NULL,
1940         PA_TAG_USEC, pa_source_get_latency(source),
1941         PA_TAG_STRING, source->driver,
1942         PA_TAG_U32, source->flags,
1943         PA_TAG_INVALID);
1944 }
1945
1946 static void client_fill_tagstruct(pa_tagstruct *t, pa_client *client) {
1947     pa_assert(t);
1948     pa_assert(client);
1949
1950     pa_tagstruct_putu32(t, client->index);
1951     pa_tagstruct_puts(t, client->name);
1952     pa_tagstruct_putu32(t, client->owner ? client->owner->index : PA_INVALID_INDEX);
1953     pa_tagstruct_puts(t, client->driver);
1954 }
1955
1956 static void module_fill_tagstruct(pa_tagstruct *t, pa_module *module) {
1957     pa_assert(t);
1958     pa_assert(module);
1959
1960     pa_tagstruct_putu32(t, module->index);
1961     pa_tagstruct_puts(t, module->name);
1962     pa_tagstruct_puts(t, module->argument);
1963     pa_tagstruct_putu32(t, module->n_used);
1964     pa_tagstruct_put_boolean(t, module->auto_unload);
1965 }
1966
1967 static void sink_input_fill_tagstruct(connection *c, pa_tagstruct *t, pa_sink_input *s) {
1968     pa_assert(t);
1969     pa_sink_input_assert_ref(s);
1970
1971     pa_tagstruct_putu32(t, s->index);
1972     pa_tagstruct_puts(t, s->name);
1973     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
1974     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
1975     pa_tagstruct_putu32(t, s->sink->index);
1976     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1977     pa_tagstruct_put_channel_map(t, &s->channel_map);
1978     pa_tagstruct_put_cvolume(t, &s->volume);
1979     pa_tagstruct_put_usec(t, pa_sink_input_get_latency(s));
1980     pa_tagstruct_put_usec(t, pa_sink_get_latency(s->sink));
1981     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
1982     pa_tagstruct_puts(t, s->driver);
1983     if (c->version >= 11)
1984         pa_tagstruct_put_boolean(t, pa_sink_input_get_mute(s));
1985 }
1986
1987 static void source_output_fill_tagstruct(pa_tagstruct *t, pa_source_output *s) {
1988     pa_assert(t);
1989     pa_source_output_assert_ref(s);
1990
1991     pa_tagstruct_putu32(t, s->index);
1992     pa_tagstruct_puts(t, s->name);
1993     pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
1994     pa_tagstruct_putu32(t, s->client ? s->client->index : PA_INVALID_INDEX);
1995     pa_tagstruct_putu32(t, s->source->index);
1996     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1997     pa_tagstruct_put_channel_map(t, &s->channel_map);
1998     pa_tagstruct_put_usec(t, pa_source_output_get_latency(s));
1999     pa_tagstruct_put_usec(t, pa_source_get_latency(s->source));
2000     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_source_output_get_resample_method(s)));
2001     pa_tagstruct_puts(t, s->driver);
2002 }
2003
2004 static void scache_fill_tagstruct(pa_tagstruct *t, pa_scache_entry *e) {
2005     pa_assert(t);
2006     pa_assert(e);
2007
2008     pa_tagstruct_putu32(t, e->index);
2009     pa_tagstruct_puts(t, e->name);
2010     pa_tagstruct_put_cvolume(t, &e->volume);
2011     pa_tagstruct_put_usec(t, pa_bytes_to_usec(e->memchunk.length, &e->sample_spec));
2012     pa_tagstruct_put_sample_spec(t, &e->sample_spec);
2013     pa_tagstruct_put_channel_map(t, &e->channel_map);
2014     pa_tagstruct_putu32(t, e->memchunk.length);
2015     pa_tagstruct_put_boolean(t, e->lazy);
2016     pa_tagstruct_puts(t, e->filename);
2017 }
2018
2019 static void command_get_info(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2020     connection *c = CONNECTION(userdata);
2021     uint32_t idx;
2022     pa_sink *sink = NULL;
2023     pa_source *source = NULL;
2024     pa_client *client = NULL;
2025     pa_module *module = NULL;
2026     pa_sink_input *si = NULL;
2027     pa_source_output *so = NULL;
2028     pa_scache_entry *sce = NULL;
2029     const char *name;
2030     pa_tagstruct *reply;
2031
2032     connection_assert_ref(c);
2033     pa_assert(t);
2034
2035     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2036         (command != PA_COMMAND_GET_CLIENT_INFO &&
2037          command != PA_COMMAND_GET_MODULE_INFO &&
2038          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
2039          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
2040          pa_tagstruct_gets(t, &name) < 0) ||
2041         !pa_tagstruct_eof(t)) {
2042         protocol_error(c);
2043         return;
2044     }
2045
2046     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2047     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2048
2049     if (command == PA_COMMAND_GET_SINK_INFO) {
2050         if (idx != PA_INVALID_INDEX)
2051             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
2052         else
2053             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2054     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
2055         if (idx != PA_INVALID_INDEX)
2056             source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
2057         else
2058             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2059     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
2060         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
2061     else if (command == PA_COMMAND_GET_MODULE_INFO)
2062         module = pa_idxset_get_by_index(c->protocol->core->modules, idx);
2063     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
2064         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2065     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
2066         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
2067     else {
2068         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO);
2069         if (idx != PA_INVALID_INDEX)
2070             sce = pa_idxset_get_by_index(c->protocol->core->scache, idx);
2071         else
2072             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE, 0);
2073     }
2074
2075     if (!sink && !source && !client && !module && !si && !so && !sce) {
2076         pa_pstream_send_error(c->pstream, tag, PA_ERR_NOENTITY);
2077         return;
2078     }
2079
2080     reply = reply_new(tag);
2081     if (sink)
2082         sink_fill_tagstruct(reply, sink);
2083     else if (source)
2084         source_fill_tagstruct(reply, source);
2085     else if (client)
2086         client_fill_tagstruct(reply, client);
2087     else if (module)
2088         module_fill_tagstruct(reply, module);
2089     else if (si)
2090         sink_input_fill_tagstruct(c, reply, si);
2091     else if (so)
2092         source_output_fill_tagstruct(reply, so);
2093     else
2094         scache_fill_tagstruct(reply, sce);
2095     pa_pstream_send_tagstruct(c->pstream, reply);
2096 }
2097
2098 static void command_get_info_list(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2099     connection *c = CONNECTION(userdata);
2100     pa_idxset *i;
2101     uint32_t idx;
2102     void *p;
2103     pa_tagstruct *reply;
2104
2105     connection_assert_ref(c);
2106     pa_assert(t);
2107
2108     if (!pa_tagstruct_eof(t)) {
2109         protocol_error(c);
2110         return;
2111     }
2112
2113     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2114
2115     reply = reply_new(tag);
2116
2117     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
2118         i = c->protocol->core->sinks;
2119     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
2120         i = c->protocol->core->sources;
2121     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
2122         i = c->protocol->core->clients;
2123     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
2124         i = c->protocol->core->modules;
2125     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
2126         i = c->protocol->core->sink_inputs;
2127     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
2128         i = c->protocol->core->source_outputs;
2129     else {
2130         pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
2131         i = c->protocol->core->scache;
2132     }
2133
2134     if (i) {
2135         for (p = pa_idxset_first(i, &idx); p; p = pa_idxset_next(i, &idx)) {
2136             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
2137                 sink_fill_tagstruct(reply, p);
2138             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
2139                 source_fill_tagstruct(reply, p);
2140             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
2141                 client_fill_tagstruct(reply, p);
2142             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
2143                 module_fill_tagstruct(reply, p);
2144             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
2145                 sink_input_fill_tagstruct(c, reply, p);
2146             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
2147                 source_output_fill_tagstruct(reply, p);
2148             else {
2149                 pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
2150                 scache_fill_tagstruct(reply, p);
2151             }
2152         }
2153     }
2154
2155     pa_pstream_send_tagstruct(c->pstream, reply);
2156 }
2157
2158 static void command_get_server_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2159     connection *c = CONNECTION(userdata);
2160     pa_tagstruct *reply;
2161     char txt[256];
2162     const char *n;
2163
2164     connection_assert_ref(c);
2165     pa_assert(t);
2166
2167     if (!pa_tagstruct_eof(t)) {
2168         protocol_error(c);
2169         return;
2170     }
2171
2172     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2173
2174     reply = reply_new(tag);
2175     pa_tagstruct_puts(reply, PACKAGE_NAME);
2176     pa_tagstruct_puts(reply, PACKAGE_VERSION);
2177     pa_tagstruct_puts(reply, pa_get_user_name(txt, sizeof(txt)));
2178     pa_tagstruct_puts(reply, pa_get_fqdn(txt, sizeof(txt)));
2179     pa_tagstruct_put_sample_spec(reply, &c->protocol->core->default_sample_spec);
2180
2181     n = pa_namereg_get_default_sink_name(c->protocol->core);
2182     pa_tagstruct_puts(reply, n);
2183     n = pa_namereg_get_default_source_name(c->protocol->core);
2184     pa_tagstruct_puts(reply, n);
2185
2186     pa_tagstruct_putu32(reply, c->protocol->core->cookie);
2187
2188     pa_pstream_send_tagstruct(c->pstream, reply);
2189 }
2190
2191 static void subscription_cb(pa_core *core, pa_subscription_event_type_t e, uint32_t idx, void *userdata) {
2192     pa_tagstruct *t;
2193     connection *c = CONNECTION(userdata);
2194
2195     connection_assert_ref(c);
2196
2197     t = pa_tagstruct_new(NULL, 0);
2198     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
2199     pa_tagstruct_putu32(t, (uint32_t) -1);
2200     pa_tagstruct_putu32(t, e);
2201     pa_tagstruct_putu32(t, idx);
2202     pa_pstream_send_tagstruct(c->pstream, t);
2203 }
2204
2205 static void command_subscribe(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2206     connection *c = CONNECTION(userdata);
2207     pa_subscription_mask_t m;
2208
2209     connection_assert_ref(c);
2210     pa_assert(t);
2211
2212     if (pa_tagstruct_getu32(t, &m) < 0 ||
2213         !pa_tagstruct_eof(t)) {
2214         protocol_error(c);
2215         return;
2216     }
2217
2218     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2219     CHECK_VALIDITY(c->pstream, (m & ~PA_SUBSCRIPTION_MASK_ALL) == 0, tag, PA_ERR_INVALID);
2220
2221     if (c->subscription)
2222         pa_subscription_free(c->subscription);
2223
2224     if (m != 0) {
2225         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
2226         pa_assert(c->subscription);
2227     } else
2228         c->subscription = NULL;
2229
2230     pa_pstream_send_simple_ack(c->pstream, tag);
2231 }
2232
2233 static void command_set_volume(
2234         PA_GCC_UNUSED pa_pdispatch *pd,
2235         uint32_t command,
2236         uint32_t tag,
2237         pa_tagstruct *t,
2238         void *userdata) {
2239
2240     connection *c = CONNECTION(userdata);
2241     uint32_t idx;
2242     pa_cvolume volume;
2243     pa_sink *sink = NULL;
2244     pa_source *source = NULL;
2245     pa_sink_input *si = NULL;
2246     const char *name = NULL;
2247
2248     connection_assert_ref(c);
2249     pa_assert(t);
2250
2251     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2252         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
2253         (command == PA_COMMAND_SET_SOURCE_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
2254         pa_tagstruct_get_cvolume(t, &volume) ||
2255         !pa_tagstruct_eof(t)) {
2256         protocol_error(c);
2257         return;
2258     }
2259
2260     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2261     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2262     CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
2263
2264     switch (command) {
2265
2266         case PA_COMMAND_SET_SINK_VOLUME:
2267             if (idx != PA_INVALID_INDEX)
2268                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
2269             else
2270                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2271             break;
2272
2273         case PA_COMMAND_SET_SOURCE_VOLUME:
2274             if (idx != PA_INVALID_INDEX)
2275                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
2276             else
2277                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2278             break;
2279
2280         case PA_COMMAND_SET_SINK_INPUT_VOLUME:
2281             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2282             break;
2283
2284         default:
2285             pa_assert_not_reached();
2286     }
2287
2288     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
2289
2290     if (sink)
2291         pa_sink_set_volume(sink, &volume);
2292     else if (source)
2293         pa_source_set_volume(source, &volume);
2294     else if (si)
2295         pa_sink_input_set_volume(si, &volume);
2296
2297     pa_pstream_send_simple_ack(c->pstream, tag);
2298 }
2299
2300 static void command_set_mute(
2301         PA_GCC_UNUSED pa_pdispatch *pd,
2302         uint32_t command,
2303         uint32_t tag,
2304         pa_tagstruct *t,
2305         void *userdata) {
2306
2307     connection *c = CONNECTION(userdata);
2308     uint32_t idx;
2309     int mute;
2310     pa_sink *sink = NULL;
2311     pa_source *source = NULL;
2312     pa_sink_input *si = NULL;
2313     const char *name = NULL;
2314
2315     connection_assert_ref(c);
2316     pa_assert(t);
2317
2318     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2319         (command == PA_COMMAND_SET_SINK_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
2320         (command == PA_COMMAND_SET_SOURCE_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
2321         pa_tagstruct_get_boolean(t, &mute) ||
2322         !pa_tagstruct_eof(t)) {
2323         protocol_error(c);
2324         return;
2325     }
2326
2327     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2328     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2329
2330     switch (command) {
2331
2332         case PA_COMMAND_SET_SINK_MUTE:
2333
2334             if (idx != PA_INVALID_INDEX)
2335                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
2336             else
2337                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2338
2339             break;
2340
2341         case PA_COMMAND_SET_SOURCE_MUTE:
2342             if (idx != PA_INVALID_INDEX)
2343                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
2344             else
2345                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2346
2347             break;
2348
2349         case PA_COMMAND_SET_SINK_INPUT_MUTE:
2350             si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2351             break;
2352
2353         default:
2354             pa_assert_not_reached();
2355     }
2356
2357     CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
2358
2359     if (sink)
2360         pa_sink_set_mute(sink, mute);
2361     else if (source)
2362         pa_source_set_mute(source, mute);
2363     else if (si)
2364         pa_sink_input_set_mute(si, mute);
2365
2366     pa_pstream_send_simple_ack(c->pstream, tag);
2367 }
2368
2369 static void command_cork_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2370     connection *c = CONNECTION(userdata);
2371     uint32_t idx;
2372     int b;
2373     playback_stream *s;
2374
2375     connection_assert_ref(c);
2376     pa_assert(t);
2377
2378     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2379         pa_tagstruct_get_boolean(t, &b) < 0 ||
2380         !pa_tagstruct_eof(t)) {
2381         protocol_error(c);
2382         return;
2383     }
2384
2385     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2386     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2387     s = pa_idxset_get_by_index(c->output_streams, idx);
2388     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2389     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2390
2391     pa_sink_input_cork(s->sink_input, b);
2392     pa_pstream_send_simple_ack(c->pstream, tag);
2393 }
2394
2395 static void command_trigger_or_flush_or_prebuf_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2396     connection *c = CONNECTION(userdata);
2397     uint32_t idx;
2398     playback_stream *s;
2399
2400     connection_assert_ref(c);
2401     pa_assert(t);
2402
2403     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2404         !pa_tagstruct_eof(t)) {
2405         protocol_error(c);
2406         return;
2407     }
2408
2409     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2410     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2411     s = pa_idxset_get_by_index(c->output_streams, idx);
2412     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2413     CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2414
2415     switch (command) {
2416         case PA_COMMAND_FLUSH_PLAYBACK_STREAM:
2417             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_FLUSH, NULL, 0, NULL);
2418             break;
2419
2420         case PA_COMMAND_PREBUF_PLAYBACK_STREAM:
2421             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_PREBUF_FORCE, NULL, 0, NULL);
2422             break;
2423
2424         case PA_COMMAND_TRIGGER_PLAYBACK_STREAM:
2425             pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_TRIGGER, NULL, 0, NULL);
2426             break;
2427
2428         default:
2429             pa_assert_not_reached();
2430     }
2431
2432     pa_pstream_send_simple_ack(c->pstream, tag);
2433 }
2434
2435 static void command_cork_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2436     connection *c = CONNECTION(userdata);
2437     uint32_t idx;
2438     record_stream *s;
2439     int b;
2440
2441     connection_assert_ref(c);
2442     pa_assert(t);
2443
2444     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2445         pa_tagstruct_get_boolean(t, &b) < 0 ||
2446         !pa_tagstruct_eof(t)) {
2447         protocol_error(c);
2448         return;
2449     }
2450
2451     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2452     s = pa_idxset_get_by_index(c->record_streams, idx);
2453     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2454
2455     pa_source_output_cork(s->source_output, b);
2456     pa_memblockq_prebuf_force(s->memblockq);
2457     pa_pstream_send_simple_ack(c->pstream, tag);
2458 }
2459
2460 static void command_flush_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2461     connection *c = CONNECTION(userdata);
2462     uint32_t idx;
2463     record_stream *s;
2464
2465     connection_assert_ref(c);
2466     pa_assert(t);
2467
2468     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2469         !pa_tagstruct_eof(t)) {
2470         protocol_error(c);
2471         return;
2472     }
2473
2474     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2475     s = pa_idxset_get_by_index(c->record_streams, idx);
2476     CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2477
2478     pa_memblockq_flush(s->memblockq);
2479     pa_pstream_send_simple_ack(c->pstream, tag);
2480 }
2481
2482 static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2483     connection *c = CONNECTION(userdata);
2484     uint32_t idx;
2485     uint32_t maxlength, tlength, prebuf, minreq, fragsize;
2486     pa_tagstruct *reply;
2487
2488     connection_assert_ref(c);
2489     pa_assert(t);
2490
2491     if (pa_tagstruct_getu32(t, &idx) < 0) {
2492         protocol_error(c);
2493         return;
2494     }
2495
2496     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2497
2498     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR) {
2499         playback_stream *s;
2500
2501         s = pa_idxset_get_by_index(c->output_streams, idx);
2502         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2503         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2504
2505         if (pa_tagstruct_get(
2506                     t,
2507                     PA_TAG_U32, &maxlength,
2508                     PA_TAG_U32, &tlength,
2509                     PA_TAG_U32, &prebuf,
2510                     PA_TAG_U32, &minreq,
2511                     PA_TAG_INVALID) < 0 ||
2512             !pa_tagstruct_eof(t)) {
2513             protocol_error(c);
2514             return;
2515         }
2516
2517         CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
2518         CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
2519
2520         pa_memblockq_set_maxlength(s->memblockq, maxlength);
2521         pa_memblockq_set_tlength(s->memblockq, tlength);
2522         pa_memblockq_set_prebuf(s->memblockq, prebuf);
2523         pa_memblockq_set_minreq(s->memblockq, minreq);
2524
2525         reply = reply_new(tag);
2526         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_maxlength(s->memblockq));
2527         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_tlength(s->memblockq));
2528         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_prebuf(s->memblockq));
2529         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_minreq(s->memblockq));
2530
2531     } else {
2532         record_stream *s;
2533         size_t base;
2534         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR);
2535
2536         s = pa_idxset_get_by_index(c->record_streams, idx);
2537         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2538
2539         if (pa_tagstruct_get(
2540                     t,
2541                     PA_TAG_U32, &maxlength,
2542                     PA_TAG_U32, &fragsize,
2543                     PA_TAG_INVALID) < 0 ||
2544             !pa_tagstruct_eof(t)) {
2545             protocol_error(c);
2546             return;
2547         }
2548
2549         CHECK_VALIDITY(c->pstream, maxlength > 0, tag, PA_ERR_INVALID);
2550         CHECK_VALIDITY(c->pstream, maxlength <= MAX_MEMBLOCKQ_LENGTH, tag, PA_ERR_INVALID);
2551
2552         pa_memblockq_set_maxlength(s->memblockq, maxlength);
2553
2554         base = pa_frame_size(&s->source_output->sample_spec);
2555         s->fragment_size = (fragsize/base)*base;
2556         if (s->fragment_size <= 0)
2557             s->fragment_size = base;
2558
2559         if (s->fragment_size > pa_memblockq_get_maxlength(s->memblockq))
2560             s->fragment_size = pa_memblockq_get_maxlength(s->memblockq);
2561
2562         reply = reply_new(tag);
2563         pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_maxlength(s->memblockq));
2564         pa_tagstruct_putu32(reply, s->fragment_size);
2565     }
2566
2567     pa_pstream_send_tagstruct(c->pstream, reply);
2568 }
2569
2570 static void command_update_stream_sample_rate(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2571     connection *c = CONNECTION(userdata);
2572     uint32_t idx;
2573     uint32_t rate;
2574
2575     connection_assert_ref(c);
2576     pa_assert(t);
2577
2578     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2579         pa_tagstruct_getu32(t, &rate) < 0 ||
2580         !pa_tagstruct_eof(t)) {
2581         protocol_error(c);
2582         return;
2583     }
2584
2585     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2586     CHECK_VALIDITY(c->pstream, rate > 0 && rate <= PA_RATE_MAX, tag, PA_ERR_INVALID);
2587
2588     if (command == PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE) {
2589         playback_stream *s;
2590
2591         s = pa_idxset_get_by_index(c->output_streams, idx);
2592         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2593         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2594
2595         pa_sink_input_set_rate(s->sink_input, rate);
2596
2597     } else {
2598         record_stream *s;
2599         pa_assert(command == PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE);
2600
2601         s = pa_idxset_get_by_index(c->record_streams, idx);
2602         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2603
2604         pa_source_output_set_rate(s->source_output, rate);
2605     }
2606
2607     pa_pstream_send_simple_ack(c->pstream, tag);
2608 }
2609
2610 static void command_set_default_sink_or_source(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2611     connection *c = CONNECTION(userdata);
2612     const char *s;
2613
2614     connection_assert_ref(c);
2615     pa_assert(t);
2616
2617     if (pa_tagstruct_gets(t, &s) < 0 ||
2618         !pa_tagstruct_eof(t)) {
2619         protocol_error(c);
2620         return;
2621     }
2622
2623     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2624     CHECK_VALIDITY(c->pstream, !s || (*s && pa_utf8_valid(s)), tag, PA_ERR_INVALID);
2625
2626     pa_namereg_set_default(c->protocol->core, s, command == PA_COMMAND_SET_DEFAULT_SOURCE ? PA_NAMEREG_SOURCE : PA_NAMEREG_SINK);
2627     pa_pstream_send_simple_ack(c->pstream, tag);
2628 }
2629
2630 static void command_set_stream_name(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2631     connection *c = CONNECTION(userdata);
2632     uint32_t idx;
2633     const char *name;
2634
2635     connection_assert_ref(c);
2636     pa_assert(t);
2637
2638     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2639         pa_tagstruct_gets(t, &name) < 0 ||
2640         !pa_tagstruct_eof(t)) {
2641         protocol_error(c);
2642         return;
2643     }
2644
2645     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2646     CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
2647
2648     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
2649         playback_stream *s;
2650
2651         s = pa_idxset_get_by_index(c->output_streams, idx);
2652         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2653         CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
2654
2655         pa_sink_input_set_name(s->sink_input, name);
2656
2657     } else {
2658         record_stream *s;
2659         pa_assert(command == PA_COMMAND_SET_RECORD_STREAM_NAME);
2660
2661         s = pa_idxset_get_by_index(c->record_streams, idx);
2662         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2663
2664         pa_source_output_set_name(s->source_output, name);
2665     }
2666
2667     pa_pstream_send_simple_ack(c->pstream, tag);
2668 }
2669
2670 static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2671     connection *c = CONNECTION(userdata);
2672     uint32_t idx;
2673
2674     connection_assert_ref(c);
2675     pa_assert(t);
2676
2677     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2678         !pa_tagstruct_eof(t)) {
2679         protocol_error(c);
2680         return;
2681     }
2682
2683     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2684
2685     if (command == PA_COMMAND_KILL_CLIENT) {
2686         pa_client *client;
2687
2688         client = pa_idxset_get_by_index(c->protocol->core->clients, idx);
2689         CHECK_VALIDITY(c->pstream, client, tag, PA_ERR_NOENTITY);
2690
2691         connection_ref(c);
2692         pa_client_kill(client);
2693
2694     } else if (command == PA_COMMAND_KILL_SINK_INPUT) {
2695         pa_sink_input *s;
2696
2697         s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2698         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2699
2700         connection_ref(c);
2701         pa_sink_input_kill(s);
2702     } else {
2703         pa_source_output *s;
2704
2705         pa_assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
2706
2707         s = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
2708         CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
2709
2710         connection_ref(c);
2711         pa_source_output_kill(s);
2712     }
2713
2714     pa_pstream_send_simple_ack(c->pstream, tag);
2715     connection_unref(c);
2716 }
2717
2718 static void command_load_module(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2719     connection *c = CONNECTION(userdata);
2720     pa_module *m;
2721     const char *name, *argument;
2722     pa_tagstruct *reply;
2723
2724     connection_assert_ref(c);
2725     pa_assert(t);
2726
2727     if (pa_tagstruct_gets(t, &name) < 0 ||
2728         pa_tagstruct_gets(t, &argument) < 0 ||
2729         !pa_tagstruct_eof(t)) {
2730         protocol_error(c);
2731         return;
2732     }
2733
2734     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2735     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name) && !strchr(name, '/'), tag, PA_ERR_INVALID);
2736     CHECK_VALIDITY(c->pstream, !argument || pa_utf8_valid(argument), tag, PA_ERR_INVALID);
2737
2738     if (!(m = pa_module_load(c->protocol->core, name, argument))) {
2739         pa_pstream_send_error(c->pstream, tag, PA_ERR_MODINITFAILED);
2740         return;
2741     }
2742
2743     reply = reply_new(tag);
2744     pa_tagstruct_putu32(reply, m->index);
2745     pa_pstream_send_tagstruct(c->pstream, reply);
2746 }
2747
2748 static void command_unload_module(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2749     connection *c = CONNECTION(userdata);
2750     uint32_t idx;
2751     pa_module *m;
2752
2753     connection_assert_ref(c);
2754     pa_assert(t);
2755
2756     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2757         !pa_tagstruct_eof(t)) {
2758         protocol_error(c);
2759         return;
2760     }
2761
2762     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2763     m = pa_idxset_get_by_index(c->protocol->core->modules, idx);
2764     CHECK_VALIDITY(c->pstream, m, tag, PA_ERR_NOENTITY);
2765
2766     pa_module_unload_request(m);
2767     pa_pstream_send_simple_ack(c->pstream, tag);
2768 }
2769
2770 static void command_add_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2771     connection *c = CONNECTION(userdata);
2772     const char *name, *module, *argument;
2773     uint32_t type;
2774     uint32_t idx;
2775     pa_tagstruct *reply;
2776
2777     connection_assert_ref(c);
2778     pa_assert(t);
2779
2780     if (pa_tagstruct_gets(t, &name) < 0 ||
2781         pa_tagstruct_getu32(t, &type) < 0 ||
2782         pa_tagstruct_gets(t, &module) < 0 ||
2783         pa_tagstruct_gets(t, &argument) < 0 ||
2784         !pa_tagstruct_eof(t)) {
2785         protocol_error(c);
2786         return;
2787     }
2788
2789     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2790     CHECK_VALIDITY(c->pstream, name && *name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
2791     CHECK_VALIDITY(c->pstream, type == 0 || type == 1, tag, PA_ERR_INVALID);
2792     CHECK_VALIDITY(c->pstream, module && *module && pa_utf8_valid(module), tag, PA_ERR_INVALID);
2793     CHECK_VALIDITY(c->pstream, !argument || pa_utf8_valid(argument), tag, PA_ERR_INVALID);
2794
2795     if (pa_autoload_add(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE, module, argument, &idx) < 0) {
2796         pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
2797         return;
2798     }
2799
2800     reply = reply_new(tag);
2801     pa_tagstruct_putu32(reply, idx);
2802     pa_pstream_send_tagstruct(c->pstream, reply);
2803 }
2804
2805 static void command_remove_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2806     connection *c = CONNECTION(userdata);
2807     const char *name = NULL;
2808     uint32_t type, idx = PA_IDXSET_INVALID;
2809     int r;
2810
2811     connection_assert_ref(c);
2812     pa_assert(t);
2813
2814     if ((pa_tagstruct_getu32(t, &idx) < 0 &&
2815         (pa_tagstruct_gets(t, &name) < 0 ||
2816          pa_tagstruct_getu32(t, &type) < 0)) ||
2817         !pa_tagstruct_eof(t)) {
2818         protocol_error(c);
2819         return;
2820     }
2821
2822     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2823     CHECK_VALIDITY(c->pstream, name || idx != PA_IDXSET_INVALID, tag, PA_ERR_INVALID);
2824     CHECK_VALIDITY(c->pstream, !name || (*name && pa_utf8_valid(name) && (type == 0 || type == 1)), tag, PA_ERR_INVALID);
2825
2826     if (name)
2827         r = pa_autoload_remove_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
2828     else
2829         r = pa_autoload_remove_by_index(c->protocol->core, idx);
2830
2831     CHECK_VALIDITY(c->pstream, r >= 0, tag, PA_ERR_NOENTITY);
2832
2833     pa_pstream_send_simple_ack(c->pstream, tag);
2834 }
2835
2836 static void autoload_fill_tagstruct(pa_tagstruct *t, const pa_autoload_entry *e) {
2837     pa_assert(t && e);
2838
2839     pa_tagstruct_putu32(t, e->index);
2840     pa_tagstruct_puts(t, e->name);
2841     pa_tagstruct_putu32(t, e->type == PA_NAMEREG_SINK ? 0 : 1);
2842     pa_tagstruct_puts(t, e->module);
2843     pa_tagstruct_puts(t, e->argument);
2844 }
2845
2846 static void command_get_autoload_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2847     connection *c = CONNECTION(userdata);
2848     const pa_autoload_entry *a = NULL;
2849     uint32_t type, idx;
2850     const char *name;
2851     pa_tagstruct *reply;
2852
2853     connection_assert_ref(c);
2854     pa_assert(t);
2855
2856     if ((pa_tagstruct_getu32(t, &idx) < 0 &&
2857         (pa_tagstruct_gets(t, &name) < 0 ||
2858          pa_tagstruct_getu32(t, &type) < 0)) ||
2859         !pa_tagstruct_eof(t)) {
2860         protocol_error(c);
2861         return;
2862     }
2863
2864     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2865     CHECK_VALIDITY(c->pstream, name || idx != PA_IDXSET_INVALID, tag, PA_ERR_INVALID);
2866     CHECK_VALIDITY(c->pstream, !name || (*name && (type == 0 || type == 1) && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2867
2868     if (name)
2869         a = pa_autoload_get_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
2870     else
2871         a = pa_autoload_get_by_index(c->protocol->core, idx);
2872
2873     CHECK_VALIDITY(c->pstream, a, tag, PA_ERR_NOENTITY);
2874
2875     reply = reply_new(tag);
2876     autoload_fill_tagstruct(reply, a);
2877     pa_pstream_send_tagstruct(c->pstream, reply);
2878 }
2879
2880 static void command_get_autoload_info_list(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2881     connection *c = CONNECTION(userdata);
2882     pa_tagstruct *reply;
2883
2884     connection_assert_ref(c);
2885     pa_assert(t);
2886
2887     if (!pa_tagstruct_eof(t)) {
2888         protocol_error(c);
2889         return;
2890     }
2891
2892     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2893
2894     reply = reply_new(tag);
2895
2896     if (c->protocol->core->autoload_hashmap) {
2897         pa_autoload_entry *a;
2898         void *state = NULL;
2899
2900         while ((a = pa_hashmap_iterate(c->protocol->core->autoload_hashmap, &state, NULL)))
2901             autoload_fill_tagstruct(reply, a);
2902     }
2903
2904     pa_pstream_send_tagstruct(c->pstream, reply);
2905 }
2906
2907 static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2908     connection *c = CONNECTION(userdata);
2909     uint32_t idx = PA_INVALID_INDEX, idx_device = PA_INVALID_INDEX;
2910     const char *name = NULL;
2911
2912     connection_assert_ref(c);
2913     pa_assert(t);
2914
2915     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2916         pa_tagstruct_getu32(t, &idx_device) < 0 ||
2917         pa_tagstruct_gets(t, &name) < 0 ||
2918         !pa_tagstruct_eof(t)) {
2919         protocol_error(c);
2920         return;
2921     }
2922
2923     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2924     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
2925     CHECK_VALIDITY(c->pstream, idx_device != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
2926
2927     if (command == PA_COMMAND_MOVE_SINK_INPUT) {
2928         pa_sink_input *si = NULL;
2929         pa_sink *sink = NULL;
2930
2931         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
2932
2933         if (idx_device != PA_INVALID_INDEX)
2934             sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx_device);
2935         else
2936             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
2937
2938         CHECK_VALIDITY(c->pstream, si && sink, tag, PA_ERR_NOENTITY);
2939
2940         if (pa_sink_input_move_to(si, sink, 0) < 0) {
2941             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
2942             return;
2943         }
2944     } else {
2945         pa_source_output *so = NULL;
2946         pa_source *source;
2947
2948         pa_assert(command == PA_COMMAND_MOVE_SOURCE_OUTPUT);
2949
2950         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
2951
2952         if (idx_device != PA_INVALID_INDEX)
2953             source = pa_idxset_get_by_index(c->protocol->core->sources, idx_device);
2954         else
2955             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
2956
2957         CHECK_VALIDITY(c->pstream, so && source, tag, PA_ERR_NOENTITY);
2958
2959         if (pa_source_output_move_to(so, source) < 0) {
2960             pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
2961             return;
2962         }
2963     }
2964
2965     pa_pstream_send_simple_ack(c->pstream, tag);
2966 }
2967
2968 static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2969     connection *c = CONNECTION(userdata);
2970     uint32_t idx = PA_INVALID_INDEX;
2971     const char *name = NULL;
2972     int b;
2973
2974     connection_assert_ref(c);
2975     pa_assert(t);
2976
2977     if (pa_tagstruct_getu32(t, &idx) < 0 ||
2978         pa_tagstruct_gets(t, &name) < 0 ||
2979         pa_tagstruct_get_boolean(t, &b) < 0 ||
2980         !pa_tagstruct_eof(t)) {
2981         protocol_error(c);
2982         return;
2983     }
2984
2985     CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
2986     CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || !*name || pa_utf8_valid(name), tag, PA_ERR_INVALID);
2987
2988     if (command == PA_COMMAND_SUSPEND_SINK) {
2989
2990         if (idx == PA_INVALID_INDEX && name && !*name) {
2991
2992             if (pa_sink_suspend_all(c->protocol->core, b) < 0) {
2993                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
2994                 return;
2995             }
2996         } else {
2997             pa_sink *sink = NULL;
2998
2999             if (idx != PA_INVALID_INDEX)
3000                 sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
3001             else
3002                 sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
3003
3004             CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
3005
3006             if (pa_sink_suspend(sink, b) < 0) {
3007                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3008                 return;
3009             }
3010         }
3011     } else {
3012
3013         pa_assert(command == PA_COMMAND_SUSPEND_SOURCE);
3014
3015         if (idx == PA_INVALID_INDEX && name && !*name) {
3016
3017             if (pa_source_suspend_all(c->protocol->core, b) < 0) {
3018                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3019                 return;
3020             }
3021
3022         } else {
3023             pa_source *source;
3024
3025             if (idx != PA_INVALID_INDEX)
3026                 source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
3027             else
3028                 source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
3029
3030             CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
3031
3032             if (pa_source_suspend(source, b) < 0) {
3033                 pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID);
3034                 return;
3035             }
3036         }
3037     }
3038
3039     pa_pstream_send_simple_ack(c->pstream, tag);
3040 }
3041
3042 /*** pstream callbacks ***/
3043
3044 static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
3045     connection *c = CONNECTION(userdata);
3046
3047     pa_assert(p);
3048     pa_assert(packet);
3049     connection_assert_ref(c);
3050
3051     if (pa_pdispatch_run(c->pdispatch, packet, creds, c) < 0) {
3052         pa_log("invalid packet.");
3053         connection_unlink(c);
3054     }
3055 }
3056
3057 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) {
3058     connection *c = CONNECTION(userdata);
3059     output_stream *stream;
3060
3061     pa_assert(p);
3062     pa_assert(chunk);
3063     connection_assert_ref(c);
3064
3065     if (!(stream = OUTPUT_STREAM(pa_idxset_get_by_index(c->output_streams, channel)))) {
3066         pa_log("client sent block for invalid stream.");
3067         /* Ignoring */
3068         return;
3069     }
3070
3071     if (playback_stream_isinstance(stream)) {
3072         playback_stream *ps = PLAYBACK_STREAM(stream);
3073
3074         if (seek != PA_SEEK_RELATIVE || offset != 0)
3075             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);
3076
3077         pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
3078
3079     } else {
3080         upload_stream *u = UPLOAD_STREAM(stream);
3081         size_t l;
3082
3083         if (!u->memchunk.memblock) {
3084             if (u->length == chunk->length) {
3085                 u->memchunk = *chunk;
3086                 pa_memblock_ref(u->memchunk.memblock);
3087                 u->length = 0;
3088             } else {
3089                 u->memchunk.memblock = pa_memblock_new(c->protocol->core->mempool, u->length);
3090                 u->memchunk.index = u->memchunk.length = 0;
3091             }
3092         }
3093
3094         pa_assert(u->memchunk.memblock);
3095
3096         l = u->length;
3097         if (l > chunk->length)
3098             l = chunk->length;
3099
3100
3101         if (l > 0) {
3102             void *src, *dst;
3103             dst = pa_memblock_acquire(u->memchunk.memblock);
3104             src = pa_memblock_acquire(chunk->memblock);
3105
3106             memcpy((uint8_t*) dst + u->memchunk.index + u->memchunk.length,
3107                    (uint8_t*) src+chunk->index, l);
3108
3109             pa_memblock_release(u->memchunk.memblock);
3110             pa_memblock_release(chunk->memblock);
3111
3112             u->memchunk.length += l;
3113             u->length -= l;
3114         }
3115     }
3116 }
3117
3118 static void pstream_die_callback(pa_pstream *p, void *userdata) {
3119     connection *c = CONNECTION(userdata);
3120
3121     pa_assert(p);
3122     connection_assert_ref(c);
3123
3124     connection_unlink(c);
3125     pa_log_info("connection died.");
3126 }
3127
3128 static void pstream_drain_callback(pa_pstream *p, void *userdata) {
3129     connection *c = CONNECTION(userdata);
3130
3131     pa_assert(p);
3132     connection_assert_ref(c);
3133
3134     send_memblock(c);
3135 }
3136
3137 static void pstream_revoke_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
3138     pa_thread_mq *q;
3139
3140     if (!(q = pa_thread_mq_get()))
3141         pa_pstream_send_revoke(p, block_id);
3142     else
3143         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_REVOKE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
3144 }
3145
3146 static void pstream_release_callback(pa_pstream *p, uint32_t block_id, void *userdata) {
3147     pa_thread_mq *q;
3148
3149     if (!(q = pa_thread_mq_get()))
3150         pa_pstream_send_release(p, block_id);
3151     else
3152         pa_asyncmsgq_post(q->outq, PA_MSGOBJECT(userdata), CONNECTION_MESSAGE_RELEASE, PA_UINT_TO_PTR(block_id), 0, NULL, NULL);
3153 }
3154
3155 /*** client callbacks ***/
3156
3157 static void client_kill_cb(pa_client *c) {
3158     pa_assert(c);
3159
3160     connection_unlink(CONNECTION(c->userdata));
3161 }
3162
3163 /*** socket server callbacks ***/
3164
3165 static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *tv, void *userdata) {
3166     connection *c = CONNECTION(userdata);
3167
3168     pa_assert(m);
3169     pa_assert(tv);
3170     connection_assert_ref(c);
3171     pa_assert(c->auth_timeout_event == e);
3172
3173     if (!c->authorized)
3174         connection_unlink(c);
3175 }
3176
3177 static void on_connection(PA_GCC_UNUSED pa_socket_server*s, pa_iochannel *io, void *userdata) {
3178     pa_protocol_native *p = userdata;
3179     connection *c;
3180     char cname[256], pname[128];
3181
3182     pa_assert(s);
3183     pa_assert(io);
3184     pa_assert(p);
3185
3186     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
3187         pa_log_warn("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
3188         pa_iochannel_free(io);
3189         return;
3190     }
3191
3192     c = pa_msgobject_new(connection);
3193     c->parent.parent.free = connection_free;
3194     c->parent.process_msg = connection_process_msg;
3195
3196     c->authorized = !!p->public;
3197
3198     if (!c->authorized && p->auth_ip_acl && pa_ip_acl_check(p->auth_ip_acl, pa_iochannel_get_recv_fd(io)) > 0) {
3199         pa_log_info("Client authenticated by IP ACL.");
3200         c->authorized = 1;
3201     }
3202
3203     if (!c->authorized) {
3204         struct timeval tv;
3205         pa_gettimeofday(&tv);
3206         tv.tv_sec += AUTH_TIMEOUT;
3207         c->auth_timeout_event = p->core->mainloop->time_new(p->core->mainloop, &tv, auth_timeout, c);
3208     } else
3209         c->auth_timeout_event = NULL;
3210
3211     c->version = 8;
3212     c->protocol = p;
3213     pa_iochannel_socket_peer_to_string(io, pname, sizeof(pname));
3214     pa_snprintf(cname, sizeof(cname), "Native client (%s)", pname);
3215     c->client = pa_client_new(p->core, __FILE__, cname);
3216     c->client->kill = client_kill_cb;
3217     c->client->userdata = c;
3218     c->client->owner = p->module;
3219
3220     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->mempool);
3221
3222     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
3223     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
3224     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
3225     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
3226     pa_pstream_set_revoke_callback(c->pstream, pstream_revoke_callback, c);
3227     pa_pstream_set_release_callback(c->pstream, pstream_release_callback, c);
3228
3229     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
3230
3231     c->record_streams = pa_idxset_new(NULL, NULL);
3232     c->output_streams = pa_idxset_new(NULL, NULL);
3233
3234     c->rrobin_index = PA_IDXSET_INVALID;
3235     c->subscription = NULL;
3236
3237     pa_idxset_put(p->connections, c, NULL);
3238
3239 #ifdef HAVE_CREDS
3240     if (pa_iochannel_creds_supported(io))
3241         pa_iochannel_creds_enable(io);
3242
3243 #endif
3244 }
3245
3246 /*** module entry points ***/
3247
3248 static int load_key(pa_protocol_native*p, const char*fn) {
3249     pa_assert(p);
3250
3251     p->auth_cookie_in_property = 0;
3252
3253     if (!fn && pa_authkey_prop_get(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0) {
3254         pa_log_info("using already loaded auth cookie.");
3255         pa_authkey_prop_ref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
3256         p->auth_cookie_in_property = 1;
3257         return 0;
3258     }
3259
3260     if (!fn)
3261         fn = PA_NATIVE_COOKIE_FILE;
3262
3263     if (pa_authkey_load_auto(fn, p->auth_cookie, sizeof(p->auth_cookie)) < 0)
3264         return -1;
3265
3266     pa_log_info("loading cookie from disk.");
3267
3268     if (pa_authkey_prop_put(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0)
3269         p->auth_cookie_in_property = 1;
3270
3271     return 0;
3272 }
3273
3274 static pa_protocol_native* protocol_new_internal(pa_core *c, pa_module *m, pa_modargs *ma) {
3275     pa_protocol_native *p;
3276     pa_bool_t public = FALSE;
3277     const char *acl;
3278
3279     pa_assert(c);
3280     pa_assert(ma);
3281
3282     if (pa_modargs_get_value_boolean(ma, "auth-anonymous", &public) < 0) {
3283         pa_log("auth-anonymous= expects a boolean argument.");
3284         return NULL;
3285     }
3286
3287     p = pa_xnew(pa_protocol_native, 1);
3288     p->core = c;
3289     p->module = m;
3290     p->public = public;
3291     p->server = NULL;
3292     p->auth_ip_acl = NULL;
3293
3294 #ifdef HAVE_CREDS
3295     {
3296         pa_bool_t a = 1;
3297         if (pa_modargs_get_value_boolean(ma, "auth-group-enabled", &a) < 0) {
3298             pa_log("auth-group-enabled= expects a boolean argument.");
3299             return NULL;
3300         }
3301         p->auth_group = a ? pa_xstrdup(pa_modargs_get_value(ma, "auth-group", c->is_system_instance ? PA_ACCESS_GROUP : NULL)) : NULL;
3302
3303         if (p->auth_group)
3304             pa_log_info("Allowing access to group '%s'.", p->auth_group);
3305     }
3306 #endif
3307
3308
3309     if ((acl = pa_modargs_get_value(ma, "auth-ip-acl", NULL))) {
3310
3311         if (!(p->auth_ip_acl = pa_ip_acl_new(acl))) {
3312             pa_log("Failed to parse IP ACL '%s'", acl);
3313             goto fail;
3314         }
3315     }
3316
3317     if (load_key(p, pa_modargs_get_value(ma, "cookie", NULL)) < 0)
3318         goto fail;
3319
3320     p->connections = pa_idxset_new(NULL, NULL);
3321
3322     return p;
3323
3324 fail:
3325 #ifdef HAVE_CREDS
3326     pa_xfree(p->auth_group);
3327 #endif
3328     if (p->auth_ip_acl)
3329         pa_ip_acl_free(p->auth_ip_acl);
3330     pa_xfree(p);
3331     return NULL;
3332 }
3333
3334 pa_protocol_native* pa_protocol_native_new(pa_core *core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
3335     char t[256];
3336     pa_protocol_native *p;
3337
3338     if (!(p = protocol_new_internal(core, m, ma)))
3339         return NULL;
3340
3341     p->server = server;
3342     pa_socket_server_set_callback(p->server, on_connection, p);
3343
3344     if (pa_socket_server_get_address(p->server, t, sizeof(t))) {
3345         pa_strlist *l;
3346         l = pa_property_get(core, PA_NATIVE_SERVER_PROPERTY_NAME);
3347         l = pa_strlist_prepend(l, t);
3348         pa_property_replace(core, PA_NATIVE_SERVER_PROPERTY_NAME, l);
3349     }
3350
3351     return p;
3352 }
3353
3354 void pa_protocol_native_free(pa_protocol_native *p) {
3355     connection *c;
3356     pa_assert(p);
3357
3358     while ((c = pa_idxset_first(p->connections, NULL)))
3359         connection_unlink(c);
3360     pa_idxset_free(p->connections, NULL, NULL);
3361
3362     if (p->server) {
3363         char t[256];
3364
3365         if (pa_socket_server_get_address(p->server, t, sizeof(t))) {
3366             pa_strlist *l;
3367             l = pa_property_get(p->core, PA_NATIVE_SERVER_PROPERTY_NAME);
3368             l = pa_strlist_remove(l, t);
3369
3370             if (l)
3371                 pa_property_replace(p->core, PA_NATIVE_SERVER_PROPERTY_NAME, l);
3372             else
3373                 pa_property_remove(p->core, PA_NATIVE_SERVER_PROPERTY_NAME);
3374         }
3375
3376         pa_socket_server_unref(p->server);
3377     }
3378
3379     if (p->auth_cookie_in_property)
3380         pa_authkey_prop_unref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
3381
3382     if (p->auth_ip_acl)
3383         pa_ip_acl_free(p->auth_ip_acl);
3384
3385 #ifdef HAVE_CREDS
3386     pa_xfree(p->auth_group);
3387 #endif
3388     pa_xfree(p);
3389 }
3390
3391 pa_protocol_native* pa_protocol_native_new_iochannel(
3392         pa_core*core,
3393         pa_iochannel *io,
3394         pa_module *m,
3395         pa_modargs *ma) {
3396
3397     pa_protocol_native *p;
3398
3399     if (!(p = protocol_new_internal(core, m, ma)))
3400         return NULL;
3401
3402     on_connection(NULL, io, p);
3403
3404     return p;
3405 }