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