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     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_sink_input_get_resample_method(s)));
1164 }
1165
1166 static void source_output_fill_tagstruct(struct pa_tagstruct *t, struct pa_source_output *s) {
1167     assert(t && s);
1168     pa_tagstruct_putu32(t, s->index);
1169     pa_tagstruct_puts(t, s->name);
1170     pa_tagstruct_putu32(t, s->owner ? s->owner->index : (uint32_t) -1);
1171     pa_tagstruct_putu32(t, s->client ? s->client->index : (uint32_t) -1);
1172     pa_tagstruct_putu32(t, s->source->index);
1173     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
1174     pa_tagstruct_put_usec(t, pa_source_output_get_latency(s));
1175     pa_tagstruct_put_usec(t, pa_source_get_latency(s->source));
1176     pa_tagstruct_puts(t, pa_resample_method_to_string(pa_source_output_get_resample_method(s)));
1177 }
1178
1179 static void scache_fill_tagstruct(struct pa_tagstruct *t, struct pa_scache_entry *e) {
1180     assert(t && e);
1181     pa_tagstruct_putu32(t, e->index);
1182     pa_tagstruct_puts(t, e->name);
1183     pa_tagstruct_putu32(t, e->volume);
1184     pa_tagstruct_put_usec(t, pa_bytes_to_usec(e->memchunk.length, &e->sample_spec));
1185     pa_tagstruct_put_sample_spec(t, &e->sample_spec);
1186     pa_tagstruct_putu32(t, e->memchunk.length);
1187     pa_tagstruct_put_boolean(t, e->lazy);
1188     pa_tagstruct_puts(t, e->filename);
1189 }
1190
1191 static void command_get_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1192     struct connection *c = userdata;
1193     uint32_t index;
1194     struct pa_sink *sink = NULL;
1195     struct pa_source *source = NULL;
1196     struct pa_client *client = NULL;
1197     struct pa_module *module = NULL;
1198     struct pa_sink_input *si = NULL;
1199     struct pa_source_output *so = NULL;
1200     struct pa_scache_entry *sce = NULL;
1201     const char *name;
1202     struct pa_tagstruct *reply;
1203     assert(c && t);
1204
1205
1206     if (pa_tagstruct_getu32(t, &index) < 0 ||
1207         (command != PA_COMMAND_GET_CLIENT_INFO &&
1208          command != PA_COMMAND_GET_MODULE_INFO &&
1209          command != PA_COMMAND_GET_SINK_INPUT_INFO &&
1210          command != PA_COMMAND_GET_SOURCE_OUTPUT_INFO &&
1211          pa_tagstruct_gets(t, &name) < 0) ||
1212         !pa_tagstruct_eof(t)) {
1213         protocol_error(c);
1214         return;
1215     }
1216
1217     if (!c->authorized) {
1218         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1219         return;
1220     }
1221
1222     if (command == PA_COMMAND_GET_SINK_INFO) {
1223         if (index != (uint32_t) -1)
1224             sink = pa_idxset_get_by_index(c->protocol->core->sinks, index);
1225         else
1226             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
1227     } else if (command == PA_COMMAND_GET_SOURCE_INFO) {
1228         if (index != (uint32_t) -1)
1229             source = pa_idxset_get_by_index(c->protocol->core->sources, index);
1230         else
1231             source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
1232     } else if (command == PA_COMMAND_GET_CLIENT_INFO)
1233         client = pa_idxset_get_by_index(c->protocol->core->clients, index);
1234     else if (command == PA_COMMAND_GET_MODULE_INFO)
1235         module = pa_idxset_get_by_index(c->protocol->core->modules, index);
1236     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO)
1237         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index);
1238     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
1239         so = pa_idxset_get_by_index(c->protocol->core->source_outputs, index);
1240     else {
1241         assert(command == PA_COMMAND_GET_SAMPLE_INFO);
1242         if (index != (uint32_t) -1)
1243             sce = pa_idxset_get_by_index(c->protocol->core->scache, index);
1244         else
1245             sce = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SAMPLE, 0);
1246     }
1247
1248     if (!sink && !source && !client && !module && !si && !so && !sce) {
1249         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1250         return;
1251     }
1252
1253     reply = pa_tagstruct_new(NULL, 0);
1254     assert(reply);
1255     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1256     pa_tagstruct_putu32(reply, tag);
1257     if (sink)
1258         sink_fill_tagstruct(reply, sink);
1259     else if (source)
1260         source_fill_tagstruct(reply, source);
1261     else if (client)
1262         client_fill_tagstruct(reply, client);
1263     else if (module)
1264         module_fill_tagstruct(reply, module);
1265     else if (si)
1266         sink_input_fill_tagstruct(reply, si);
1267     else if (so)
1268         source_output_fill_tagstruct(reply, so);
1269     else
1270         scache_fill_tagstruct(reply, sce);
1271     pa_pstream_send_tagstruct(c->pstream, reply);
1272 }
1273
1274 static void command_get_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1275     struct connection *c = userdata;
1276     struct pa_idxset *i;
1277     uint32_t index;
1278     void *p;
1279     struct pa_tagstruct *reply;
1280     assert(c && t);
1281
1282     if (!pa_tagstruct_eof(t)) {
1283         protocol_error(c);
1284         return;
1285     }
1286
1287     if (!c->authorized) {
1288         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1289         return;
1290     }
1291
1292     reply = pa_tagstruct_new(NULL, 0);
1293     assert(reply);
1294     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1295     pa_tagstruct_putu32(reply, tag);
1296
1297     if (command == PA_COMMAND_GET_SINK_INFO_LIST)
1298         i = c->protocol->core->sinks;
1299     else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
1300         i = c->protocol->core->sources;
1301     else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
1302         i = c->protocol->core->clients;
1303     else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
1304         i = c->protocol->core->modules;
1305     else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
1306         i = c->protocol->core->sink_inputs;
1307     else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
1308         i = c->protocol->core->source_outputs;
1309     else {
1310         assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
1311         i = c->protocol->core->scache;
1312     }
1313
1314     if (i) {
1315         for (p = pa_idxset_first(i, &index); p; p = pa_idxset_next(i, &index)) {
1316             if (command == PA_COMMAND_GET_SINK_INFO_LIST)
1317                 sink_fill_tagstruct(reply, p);
1318             else if (command == PA_COMMAND_GET_SOURCE_INFO_LIST)
1319                 source_fill_tagstruct(reply, p);
1320             else if (command == PA_COMMAND_GET_CLIENT_INFO_LIST)
1321                 client_fill_tagstruct(reply, p);
1322             else if (command == PA_COMMAND_GET_MODULE_INFO_LIST)
1323                 module_fill_tagstruct(reply, p);
1324             else if (command == PA_COMMAND_GET_SINK_INPUT_INFO_LIST)
1325                 sink_input_fill_tagstruct(reply, p);
1326             else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
1327                 source_output_fill_tagstruct(reply, p);
1328             else {
1329                 assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
1330                 scache_fill_tagstruct(reply, p);
1331             }
1332         }
1333     }
1334
1335     pa_pstream_send_tagstruct(c->pstream, reply);
1336 }
1337
1338 static void command_get_server_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1339     struct connection *c = userdata;
1340     struct pa_tagstruct *reply;
1341     char txt[256];
1342     const char *n;
1343     assert(c && t);
1344
1345     if (!pa_tagstruct_eof(t)) {
1346         protocol_error(c);
1347         return;
1348     }
1349
1350     if (!c->authorized) {
1351         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1352         return;
1353     }
1354
1355     reply = pa_tagstruct_new(NULL, 0);
1356     assert(reply);
1357     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1358     pa_tagstruct_putu32(reply, tag);
1359     pa_tagstruct_puts(reply, PACKAGE_NAME);
1360     pa_tagstruct_puts(reply, PACKAGE_VERSION);
1361     pa_tagstruct_puts(reply, pa_get_user_name(txt, sizeof(txt)));
1362     pa_tagstruct_puts(reply, pa_get_fqdn(txt, sizeof(txt)));
1363     pa_tagstruct_put_sample_spec(reply, &c->protocol->core->default_sample_spec);
1364
1365     n = pa_namereg_get_default_sink_name(c->protocol->core);
1366     pa_tagstruct_puts(reply, n);
1367     n = pa_namereg_get_default_source_name(c->protocol->core);
1368     pa_tagstruct_puts(reply, n);
1369     pa_pstream_send_tagstruct(c->pstream, reply);
1370 }
1371
1372 static void subscription_cb(struct pa_core *core, enum pa_subscription_event_type e, uint32_t index, void *userdata) {
1373     struct pa_tagstruct *t;
1374     struct connection *c = userdata;
1375     assert(c && core);
1376
1377     t = pa_tagstruct_new(NULL, 0);
1378     assert(t);
1379     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
1380     pa_tagstruct_putu32(t, (uint32_t) -1);
1381     pa_tagstruct_putu32(t, e);
1382     pa_tagstruct_putu32(t, index);
1383     pa_pstream_send_tagstruct(c->pstream, t);
1384 }
1385
1386 static void command_subscribe(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1387     struct connection *c = userdata;
1388     enum pa_subscription_mask m;
1389     assert(c && t);
1390
1391     if (pa_tagstruct_getu32(t, &m) < 0 ||
1392         !pa_tagstruct_eof(t)) {
1393         protocol_error(c);
1394         return;
1395     }
1396
1397     if (!c->authorized) {
1398         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1399         return;
1400     }
1401
1402     if (c->subscription)
1403         pa_subscription_free(c->subscription);
1404
1405     if (m != 0) {
1406         c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
1407         assert(c->subscription);
1408     } else
1409         c->subscription = NULL;
1410
1411     pa_pstream_send_simple_ack(c->pstream, tag);
1412 }
1413
1414 static void command_set_volume(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1415     struct connection *c = userdata;
1416     uint32_t index, volume;
1417     struct pa_sink *sink = NULL;
1418     struct pa_sink_input *si = NULL;
1419     const char *name = NULL;
1420     assert(c && t);
1421
1422     if (pa_tagstruct_getu32(t, &index) < 0 ||
1423         (command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
1424         pa_tagstruct_getu32(t, &volume) ||
1425         !pa_tagstruct_eof(t)) {
1426         protocol_error(c);
1427         return;
1428     }
1429
1430     if (!c->authorized) {
1431         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1432         return;
1433     }
1434
1435     if (command == PA_COMMAND_SET_SINK_VOLUME) {
1436         if (index != (uint32_t) -1)
1437             sink = pa_idxset_get_by_index(c->protocol->core->sinks, index);
1438         else
1439             sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
1440     }  else {
1441         assert(command == PA_COMMAND_SET_SINK_INPUT_VOLUME);
1442         si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index);
1443     }
1444
1445     if (!si && !sink) {
1446         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1447         return;
1448     }
1449
1450     if (sink)
1451         pa_sink_set_volume(sink, volume);
1452     else if (si)
1453         pa_sink_input_set_volume(si, volume);
1454
1455     pa_pstream_send_simple_ack(c->pstream, tag);
1456 }
1457
1458 static void command_cork_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1459     struct connection *c = userdata;
1460     uint32_t index;
1461     int b;
1462     struct playback_stream *s;
1463     assert(c && t);
1464
1465     if (pa_tagstruct_getu32(t, &index) < 0 ||
1466         pa_tagstruct_get_boolean(t, &b) < 0 ||
1467         !pa_tagstruct_eof(t)) {
1468         protocol_error(c);
1469         return;
1470     }
1471
1472     if (!c->authorized) {
1473         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1474         return;
1475     }
1476
1477     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1478         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1479         return;
1480     }
1481
1482     pa_sink_input_cork(s->sink_input, b);
1483     pa_pstream_send_simple_ack(c->pstream, tag);
1484 }
1485
1486 static void command_flush_or_trigger_playback_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1487     struct connection *c = userdata;
1488     uint32_t index;
1489     struct playback_stream *s;
1490     assert(c && t);
1491
1492     if (pa_tagstruct_getu32(t, &index) < 0 ||
1493         !pa_tagstruct_eof(t)) {
1494         protocol_error(c);
1495         return;
1496     }
1497
1498     if (!c->authorized) {
1499         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1500         return;
1501     }
1502
1503     if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1504         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1505         return;
1506     }
1507
1508     if (command == PA_COMMAND_PREBUF_PLAYBACK_STREAM)
1509         pa_memblockq_prebuf_reenable(s->memblockq);
1510     else if (command == PA_COMMAND_TRIGGER_PLAYBACK_STREAM)
1511         pa_memblockq_prebuf_disable(s->memblockq);
1512     else {
1513         assert(command == PA_COMMAND_FLUSH_PLAYBACK_STREAM);
1514         pa_memblockq_flush(s->memblockq);
1515         /*pa_log(__FILE__": flush: %u\n", pa_memblockq_get_length(s->memblockq));*/
1516     }
1517
1518     pa_sink_notify(s->sink_input->sink);
1519     pa_pstream_send_simple_ack(c->pstream, tag);
1520     request_bytes(s);
1521 }
1522
1523 static void command_cork_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1524     struct connection *c = userdata;
1525     uint32_t index;
1526     struct record_stream *s;
1527     int b;
1528     assert(c && t);
1529
1530     if (pa_tagstruct_getu32(t, &index) < 0 ||
1531         pa_tagstruct_get_boolean(t, &b) < 0 ||
1532         !pa_tagstruct_eof(t)) {
1533         protocol_error(c);
1534         return;
1535     }
1536
1537     if (!c->authorized) {
1538         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1539         return;
1540     }
1541
1542     if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
1543         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1544         return;
1545     }
1546
1547     pa_source_output_cork(s->source_output, b);
1548     pa_pstream_send_simple_ack(c->pstream, tag);
1549 }
1550
1551 static void command_flush_record_stream(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1552     struct connection *c = userdata;
1553     uint32_t index;
1554     struct record_stream *s;
1555     assert(c && t);
1556
1557     if (pa_tagstruct_getu32(t, &index) < 0 ||
1558         !pa_tagstruct_eof(t)) {
1559         protocol_error(c);
1560         return;
1561     }
1562
1563     if (!c->authorized) {
1564         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1565         return;
1566     }
1567
1568     if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
1569         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1570         return;
1571     }
1572
1573     pa_memblockq_flush(s->memblockq);
1574     pa_pstream_send_simple_ack(c->pstream, tag);
1575 }
1576
1577 static void command_set_default_sink_or_source(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1578     struct connection *c = userdata;
1579     uint32_t index;
1580     const char *s;
1581     assert(c && t);
1582
1583     if (pa_tagstruct_getu32(t, &index) < 0 ||
1584         pa_tagstruct_gets(t, &s) < 0 || !s ||
1585         !pa_tagstruct_eof(t)) {
1586         protocol_error(c);
1587         return;
1588     }
1589
1590     if (!c->authorized) {
1591         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1592         return;
1593     }
1594
1595     pa_namereg_set_default(c->protocol->core, s, command == PA_COMMAND_SET_DEFAULT_SOURCE ? PA_NAMEREG_SOURCE : PA_NAMEREG_SINK);
1596     pa_pstream_send_simple_ack(c->pstream, tag);
1597 }
1598
1599 static void command_set_stream_name(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1600     struct connection *c = userdata;
1601     uint32_t index;
1602     const char *name;
1603     assert(c && t);
1604
1605     if (pa_tagstruct_getu32(t, &index) < 0 ||
1606         pa_tagstruct_gets(t, &name) < 0 || !name ||
1607         !pa_tagstruct_eof(t)) {
1608         protocol_error(c);
1609         return;
1610     }
1611
1612     if (!c->authorized) {
1613         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1614         return;
1615     }
1616
1617     if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
1618         struct playback_stream *s;
1619
1620         if (!(s = pa_idxset_get_by_index(c->output_streams, index)) || s->type != PLAYBACK_STREAM) {
1621             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1622             return;
1623         }
1624
1625         pa_sink_input_set_name(s->sink_input, name);
1626
1627     } else {
1628         struct record_stream *s;
1629
1630         if (!(s = pa_idxset_get_by_index(c->record_streams, index))) {
1631             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1632             return;
1633         }
1634
1635         pa_source_output_set_name(s->source_output, name);
1636     }
1637
1638     pa_pstream_send_simple_ack(c->pstream, tag);
1639 }
1640
1641 static void command_kill(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1642     struct connection *c = userdata;
1643     uint32_t index;
1644     assert(c && t);
1645
1646     if (pa_tagstruct_getu32(t, &index) < 0 ||
1647         !pa_tagstruct_eof(t)) {
1648         protocol_error(c);
1649         return;
1650     }
1651
1652     if (!c->authorized) {
1653         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1654         return;
1655     }
1656
1657     if (command == PA_COMMAND_KILL_CLIENT) {
1658         struct pa_client *client;
1659
1660         if (!(client = pa_idxset_get_by_index(c->protocol->core->clients, index))) {
1661             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1662             return;
1663         }
1664
1665         pa_client_kill(client);
1666     } else if (command == PA_COMMAND_KILL_SINK_INPUT) {
1667         struct pa_sink_input *s;
1668
1669         if (!(s = pa_idxset_get_by_index(c->protocol->core->sink_inputs, index))) {
1670             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1671             return;
1672         }
1673
1674         pa_sink_input_kill(s);
1675     } else {
1676         struct pa_source_output *s;
1677
1678         assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
1679
1680         if (!(s = pa_idxset_get_by_index(c->protocol->core->source_outputs, index))) {
1681             pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1682             return;
1683         }
1684
1685         pa_source_output_kill(s);
1686     }
1687
1688     pa_pstream_send_simple_ack(c->pstream, tag);
1689 }
1690
1691 static void command_load_module(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1692     struct connection *c = userdata;
1693     struct pa_module *m;
1694     const char *name, *argument;
1695     struct pa_tagstruct *reply;
1696     assert(c && t);
1697
1698     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
1699         pa_tagstruct_gets(t, &argument) < 0 ||
1700         !pa_tagstruct_eof(t)) {
1701         protocol_error(c);
1702         return;
1703     }
1704
1705     if (!c->authorized) {
1706         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1707         return;
1708     }
1709
1710     if (!(m = pa_module_load(c->protocol->core, name, argument))) {
1711         pa_pstream_send_error(c->pstream, tag, PA_ERROR_INITFAILED);
1712         return;
1713     }
1714
1715     reply = pa_tagstruct_new(NULL, 0);
1716     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1717     pa_tagstruct_putu32(reply, tag);
1718     pa_tagstruct_putu32(reply, m->index);
1719     pa_pstream_send_tagstruct(c->pstream, reply);
1720 }
1721
1722 static void command_unload_module(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1723     struct connection *c = userdata;
1724     uint32_t index;
1725     struct pa_module *m;
1726     assert(c && t);
1727
1728     if (pa_tagstruct_getu32(t, &index) < 0 ||
1729         !pa_tagstruct_eof(t)) {
1730         protocol_error(c);
1731         return;
1732     }
1733
1734     if (!c->authorized) {
1735         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1736         return;
1737     }
1738
1739     if (!(m = pa_idxset_get_by_index(c->protocol->core->modules, index))) {
1740         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1741         return;
1742     }
1743
1744     pa_module_unload_request(m);
1745     pa_pstream_send_simple_ack(c->pstream, tag);
1746 }
1747
1748 static void command_add_autoload(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1749     struct connection *c = userdata;
1750     const char *name, *module, *argument;
1751     uint32_t type;
1752     uint32_t index;
1753     struct pa_tagstruct *reply;
1754     assert(c && t);
1755
1756     if (pa_tagstruct_gets(t, &name) < 0 || !name ||
1757         pa_tagstruct_getu32(t, &type) < 0 || type > 1 ||
1758         pa_tagstruct_gets(t, &module) < 0 || !module ||
1759         pa_tagstruct_gets(t, &argument) < 0 ||
1760         !pa_tagstruct_eof(t)) {
1761         protocol_error(c);
1762         return;
1763     }
1764
1765     if (!c->authorized) {
1766         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1767         return;
1768     }
1769
1770     if (pa_autoload_add(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE, module, argument, &index) < 0) {
1771         pa_pstream_send_error(c->pstream, tag, PA_ERROR_EXIST);
1772         return;
1773     }
1774
1775     reply = pa_tagstruct_new(NULL, 0);
1776     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1777     pa_tagstruct_putu32(reply, tag);
1778     pa_tagstruct_putu32(reply, index);
1779     pa_pstream_send_tagstruct(c->pstream, reply);
1780 }
1781
1782 static void command_remove_autoload(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1783     struct connection *c = userdata;
1784     const char *name = NULL;
1785     uint32_t type, index = PA_IDXSET_INVALID;
1786     int r;
1787     assert(c && t);
1788
1789     if ((pa_tagstruct_getu32(t, &index) < 0 &&
1790         (pa_tagstruct_gets(t, &name) < 0 ||
1791          pa_tagstruct_getu32(t, &type) < 0)) ||
1792         (!name && index == PA_IDXSET_INVALID) ||
1793         (name && type > 1) ||
1794         !pa_tagstruct_eof(t)) {
1795         protocol_error(c);
1796         return;
1797     }
1798
1799     if (!c->authorized) {
1800         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1801         return;
1802     }
1803
1804     if (name)
1805         r = pa_autoload_remove_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
1806     else
1807         r = pa_autoload_remove_by_index(c->protocol->core, index);
1808
1809     if (r < 0) {
1810         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1811         return;
1812     }
1813
1814     pa_pstream_send_simple_ack(c->pstream, tag);
1815 }
1816
1817 static void autoload_fill_tagstruct(struct pa_tagstruct *t, const struct pa_autoload_entry *e) {
1818     assert(t && e);
1819
1820     pa_tagstruct_putu32(t, e->index);
1821     pa_tagstruct_puts(t, e->name);
1822     pa_tagstruct_putu32(t, e->type == PA_NAMEREG_SINK ? 0 : 1);
1823     pa_tagstruct_puts(t, e->module);
1824     pa_tagstruct_puts(t, e->argument);
1825 }
1826
1827 static void command_get_autoload_info(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1828     struct connection *c = userdata;
1829     const struct pa_autoload_entry *a = NULL;
1830     uint32_t type, index;
1831     const char *name;
1832     struct pa_tagstruct *reply;
1833     assert(c && t);
1834
1835     if ((pa_tagstruct_getu32(t, &index) < 0 &&
1836         (pa_tagstruct_gets(t, &name) < 0 ||
1837          pa_tagstruct_getu32(t, &type) < 0)) ||
1838         (!name && index == PA_IDXSET_INVALID) ||
1839         (name && type > 1) ||
1840         !pa_tagstruct_eof(t)) {
1841         protocol_error(c);
1842         return;
1843     }
1844
1845     if (!c->authorized) {
1846         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1847         return;
1848     }
1849
1850
1851     if (name)
1852         a = pa_autoload_get_by_name(c->protocol->core, name, type == 0 ? PA_NAMEREG_SINK : PA_NAMEREG_SOURCE);
1853     else
1854         a = pa_autoload_get_by_index(c->protocol->core, index);
1855
1856     if (!a) {
1857         pa_pstream_send_error(c->pstream, tag, PA_ERROR_NOENTITY);
1858         return;
1859     }
1860
1861     reply = pa_tagstruct_new(NULL, 0);
1862     assert(reply);
1863     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1864     pa_tagstruct_putu32(reply, tag);
1865     autoload_fill_tagstruct(reply, a);
1866     pa_pstream_send_tagstruct(c->pstream, reply);
1867 }
1868
1869 static void command_get_autoload_info_list(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
1870     struct connection *c = userdata;
1871     struct pa_tagstruct *reply;
1872     assert(c && t);
1873
1874     if (!pa_tagstruct_eof(t)) {
1875         protocol_error(c);
1876         return;
1877     }
1878
1879     if (!c->authorized) {
1880         pa_pstream_send_error(c->pstream, tag, PA_ERROR_ACCESS);
1881         return;
1882     }
1883
1884     reply = pa_tagstruct_new(NULL, 0);
1885     pa_tagstruct_putu32(reply, PA_COMMAND_REPLY);
1886     pa_tagstruct_putu32(reply, tag);
1887
1888     if (c->protocol->core->autoload_hashmap) {
1889         struct pa_autoload_entry *a;
1890         void *state = NULL;
1891
1892         while ((a = pa_hashmap_iterate(c->protocol->core->autoload_hashmap, &state, NULL)))
1893             autoload_fill_tagstruct(reply, a);
1894     }
1895
1896     pa_pstream_send_tagstruct(c->pstream, reply);
1897 }
1898
1899 /*** pstream callbacks ***/
1900
1901 static void pstream_packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
1902     struct connection *c = userdata;
1903     assert(p && packet && packet->data && c);
1904
1905     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
1906         pa_log(__FILE__": invalid packet.\n");
1907         connection_free(c);
1908     }
1909 }
1910
1911 static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, uint32_t delta, const struct pa_memchunk *chunk, void *userdata) {
1912     struct connection *c = userdata;
1913     struct output_stream *stream;
1914     assert(p && chunk && userdata);
1915
1916     if (!(stream = pa_idxset_get_by_index(c->output_streams, channel))) {
1917         pa_log(__FILE__": client sent block for invalid stream.\n");
1918         connection_free(c);
1919         return;
1920     }
1921
1922     if (stream->type == PLAYBACK_STREAM) {
1923         struct playback_stream *p = (struct playback_stream*) stream;
1924         if (chunk->length >= p->requested_bytes)
1925             p->requested_bytes = 0;
1926         else
1927             p->requested_bytes -= chunk->length;
1928
1929         pa_memblockq_push_align(p->memblockq, chunk, delta);
1930         assert(p->sink_input);
1931 /*         pa_log(__FILE__": after_recv: %u\n", pa_memblockq_get_length(p->memblockq)); */
1932
1933         pa_sink_notify(p->sink_input->sink);
1934 /*          pa_log(__FILE__": Recieved %u bytes.\n", chunk->length);  */
1935
1936     } else {
1937         struct upload_stream *u = (struct upload_stream*) stream;
1938         size_t l;
1939         assert(u->type == UPLOAD_STREAM);
1940
1941         if (!u->memchunk.memblock) {
1942             if (u->length == chunk->length) {
1943                 u->memchunk = *chunk;
1944                 pa_memblock_ref(u->memchunk.memblock);
1945                 u->length = 0;
1946             } else {
1947                 u->memchunk.memblock = pa_memblock_new(u->length, c->protocol->core->memblock_stat);
1948                 u->memchunk.index = u->memchunk.length = 0;
1949             }
1950         }
1951
1952         assert(u->memchunk.memblock);
1953
1954         l = u->length;
1955         if (l > chunk->length)
1956             l = chunk->length;
1957
1958         if (l > 0) {
1959             memcpy((uint8_t*) u->memchunk.memblock->data + u->memchunk.index + u->memchunk.length,
1960                    (uint8_t*) chunk->memblock->data+chunk->index, l);
1961             u->memchunk.length += l;
1962             u->length -= l;
1963         }
1964     }
1965 }
1966
1967 static void pstream_die_callback(struct pa_pstream *p, void *userdata) {
1968     struct connection *c = userdata;
1969     assert(p && c);
1970     connection_free(c);
1971
1972 /*    pa_log(__FILE__": connection died.\n");*/
1973 }
1974
1975
1976 static void pstream_drain_callback(struct pa_pstream *p, void *userdata) {
1977     struct connection *c = userdata;
1978     assert(p && c);
1979
1980     send_memblock(c);
1981 }
1982
1983 /*** client callbacks ***/
1984
1985 static void client_kill_cb(struct pa_client *c) {
1986     assert(c && c->userdata);
1987     connection_free(c->userdata);
1988 }
1989
1990 /*** socket server callbacks ***/
1991
1992 static void auth_timeout(struct pa_mainloop_api*m, struct pa_time_event *e, const struct timeval *tv, void *userdata) {
1993     struct connection *c = userdata;
1994     assert(m && tv && c && c->auth_timeout_event == e);
1995
1996     if (!c->authorized)
1997         connection_free(c);
1998 }
1999
2000 static void on_connection(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata) {
2001     struct pa_protocol_native *p = userdata;
2002     struct connection *c;
2003     assert(io && p);
2004
2005     if (pa_idxset_ncontents(p->connections)+1 > MAX_CONNECTIONS) {
2006         pa_log_warn(__FILE__": Warning! Too many connections (%u), dropping incoming connection.\n", MAX_CONNECTIONS);
2007         pa_iochannel_free(io);
2008         return;
2009     }
2010
2011     c = pa_xmalloc(sizeof(struct connection));
2012
2013     c->authorized =!! p->public;
2014
2015     if (!c->authorized) {
2016         struct timeval tv;
2017         gettimeofday(&tv, NULL);
2018         tv.tv_sec += AUTH_TIMEOUT;
2019         c->auth_timeout_event = p->core->mainloop->time_new(p->core->mainloop, &tv, auth_timeout, c);
2020     } else
2021         c->auth_timeout_event = NULL;
2022
2023     c->protocol = p;
2024     assert(p->core);
2025     c->client = pa_client_new(p->core, "NATIVE", "Client");
2026     assert(c->client);
2027     c->client->kill = client_kill_cb;
2028     c->client->userdata = c;
2029     c->client->owner = p->module;
2030
2031     c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->memblock_stat);
2032     assert(c->pstream);
2033
2034     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
2035     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
2036     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
2037     pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
2038
2039     c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
2040     assert(c->pdispatch);
2041
2042     c->record_streams = pa_idxset_new(NULL, NULL);
2043     c->output_streams = pa_idxset_new(NULL, NULL);
2044     assert(c->record_streams && c->output_streams);
2045
2046     c->rrobin_index = PA_IDXSET_INVALID;
2047     c->subscription = NULL;
2048
2049     pa_idxset_put(p->connections, c, NULL);
2050 }
2051
2052 /*** module entry points ***/
2053
2054 static int load_key(struct pa_protocol_native*p, const char*fn) {
2055     assert(p);
2056
2057     p->auth_cookie_in_property = 0;
2058
2059     if (!fn && pa_authkey_prop_get(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0) {
2060         pa_log_info(__FILE__": using already loaded auth cookie.\n");
2061         pa_authkey_prop_ref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
2062         p->auth_cookie_in_property = 1;
2063         return 0;
2064     }
2065
2066     if (!fn)
2067         fn = PA_NATIVE_COOKIE_FILE;
2068
2069     if (pa_authkey_load_auto(fn, p->auth_cookie, sizeof(p->auth_cookie)) < 0)
2070         return -1;
2071
2072     pa_log_info(__FILE__": loading cookie from disk.\n");
2073
2074     if (pa_authkey_prop_put(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME, p->auth_cookie, sizeof(p->auth_cookie)) >= 0)
2075         p->auth_cookie_in_property = 1;
2076
2077     return 0;
2078 }
2079
2080 static struct pa_protocol_native* protocol_new_internal(struct pa_core *c, struct pa_module *m, struct pa_modargs *ma) {
2081     struct pa_protocol_native *p;
2082     int public = 0;
2083     assert(c && ma);
2084
2085     if (pa_modargs_get_value_boolean(ma, "public", &public) < 0) {
2086         pa_log(__FILE__": public= expects a boolean argument.\n");
2087         return NULL;
2088     }
2089
2090     p = pa_xmalloc(sizeof(struct pa_protocol_native));
2091     p->core = c;
2092     p->module = m;
2093     p->public = public;
2094     p->server = NULL;
2095
2096     if (load_key(p, pa_modargs_get_value(ma, "cookie", NULL)) < 0) {
2097         pa_xfree(p);
2098         return NULL;
2099     }
2100
2101     p->connections = pa_idxset_new(NULL, NULL);
2102     assert(p->connections);
2103
2104     return p;
2105 }
2106
2107 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) {
2108     char t[256];
2109     struct pa_protocol_native *p;
2110
2111     if (!(p = protocol_new_internal(core, m, ma)))
2112         return NULL;
2113
2114     p->server = server;
2115     pa_socket_server_set_callback(p->server, on_connection, p);
2116
2117     if (pa_socket_server_get_address(p->server, t, sizeof(t))) {
2118         struct pa_strlist *l;
2119         l = pa_property_get(core, PA_NATIVE_SERVER_PROPERTY_NAME);
2120         l = pa_strlist_prepend(l, t);
2121         pa_property_replace(core, PA_NATIVE_SERVER_PROPERTY_NAME, l);
2122     }
2123
2124     return p;
2125 }
2126
2127 void pa_protocol_native_free(struct pa_protocol_native *p) {
2128     struct connection *c;
2129     assert(p);
2130
2131     while ((c = pa_idxset_first(p->connections, NULL)))
2132         connection_free(c);
2133     pa_idxset_free(p->connections, NULL, NULL);
2134
2135     if (p->server) {
2136         char t[256];
2137
2138         if (pa_socket_server_get_address(p->server, t, sizeof(t))) {
2139             struct pa_strlist *l;
2140             l = pa_property_get(p->core, PA_NATIVE_SERVER_PROPERTY_NAME);
2141             l = pa_strlist_remove(l, t);
2142
2143             if (l)
2144                 pa_property_replace(p->core, PA_NATIVE_SERVER_PROPERTY_NAME, l);
2145             else
2146                 pa_property_remove(p->core, PA_NATIVE_SERVER_PROPERTY_NAME);
2147         }
2148
2149         pa_socket_server_unref(p->server);
2150     }
2151
2152     if (p->auth_cookie_in_property)
2153         pa_authkey_prop_unref(p->core, PA_NATIVE_COOKIE_PROPERTY_NAME);
2154
2155     pa_xfree(p);
2156 }
2157
2158 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) {
2159     struct pa_protocol_native *p;
2160
2161     if (!(p = protocol_new_internal(core, m, ma)))
2162         return NULL;
2163
2164     on_connection(NULL, io, p);
2165
2166     return p;
2167 }