src/polypcore/protocol-native.c

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