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