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