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