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