src/pulsecore/protocol-esound.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of PulseAudio.
   5
   6   Copyright 2004-2006 Lennart Poettering
   7   Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
   8
   9   PulseAudio is free software; you can redistribute it and/or modify
  10   it under the terms of the GNU Lesser General Public License as published
  11   by the Free Software Foundation; either version 2 of the License,
  12   or (at your option) any later version.
  13
  14   PulseAudio is distributed in the hope that it will be useful, but
  15   WITHOUT ANY WARRANTY; without even the implied warranty of
  16   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  17   General Public License for more details.
  18
  19   You should have received a copy of the GNU Lesser General Public License
  20   along with PulseAudio; if not, write to the Free Software
  21   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  22   USA.
  23 ***/
  24
  25 #ifdef HAVE_CONFIG_H
  26 #include <config.h>
  27 #endif
  28
  29 #include <errno.h>
  30 #include <string.h>
  31 #include <stdio.h>
  32 #include <stdlib.h>
  33 #include <limits.h>
  34
  35 #include <pulse/sample.h>
  36 #include <pulse/timeval.h>
  37 #include <pulse/utf8.h>
  38 #include <pulse/xmalloc.h>
  39
  40 #include <pulsecore/esound.h>
  41 #include <pulsecore/memblock.h>
  42 #include <pulsecore/client.h>
  43 #include <pulsecore/sink-input.h>
  44 #include <pulsecore/sink.h>
  45 #include <pulsecore/source-output.h>
  46 #include <pulsecore/source.h>
  47 #include <pulsecore/core-scache.h>
  48 #include <pulsecore/sample-util.h>
  49 #include <pulsecore/authkey.h>
  50 #include <pulsecore/namereg.h>
  51 #include <pulsecore/log.h>
  52 #include <pulsecore/core-util.h>
  53 #include <pulsecore/core-error.h>
  54 #include <pulsecore/ipacl.h>
  55 #include <pulsecore/macro.h>
  56 #include <pulsecore/thread-mq.h>
  57
  58 #include "endianmacros.h"
  59
  60 #include "protocol-esound.h"
  61
  62 /* Don't accept more connection than this */
  63 #define MAX_CONNECTIONS 64
  64
  65 /* Kick a client if it doesn't authenticate within this time */
  66 #define AUTH_TIMEOUT 5
  67
  68 #define DEFAULT_COOKIE_FILE ".esd_auth"
  69
  70 #define PLAYBACK_BUFFER_SECONDS (.25)
  71 #define PLAYBACK_BUFFER_FRAGMENTS (10)
  72 #define RECORD_BUFFER_SECONDS (5)
  73
  74 #define MAX_CACHE_SAMPLE_SIZE (2048000)
  75
  76 #define DEFAULT_SINK_LATENCY (150*PA_USEC_PER_MSEC)
  77 #define DEFAULT_SOURCE_LATENCY (150*PA_USEC_PER_MSEC)
  78
  79 #define SCACHE_PREFIX "esound."
  80
  81 /* This is heavily based on esound's code */
  82
  83 typedef struct connection {
  84     pa_msgobject parent;
  85
  86     uint32_t index;
  87     pa_bool_t dead;
  88     pa_protocol_esound *protocol;
  89     pa_iochannel *io;
  90     pa_client *client;
  91     pa_bool_t authorized, swap_byte_order;
  92     void *write_data;
  93     size_t write_data_alloc, write_data_index, write_data_length;
  94     void *read_data;
  95     size_t read_data_alloc, read_data_length;
  96     esd_proto_t request;
  97     esd_client_state_t state;
  98     pa_sink_input *sink_input;
  99     pa_source_output *source_output;
 100     pa_memblockq *input_memblockq, *output_memblockq;
 101     pa_defer_event *defer_event;
 102
 103     char *original_name;
 104
 105     struct {
 106         pa_memblock *current_memblock;
 107         size_t memblock_index;
 108         pa_atomic_t missing;
 109         pa_bool_t underrun;
 110     } playback;
 111
 112     struct {
 113         pa_memchunk memchunk;
 114         char *name;
 115         pa_sample_spec sample_spec;
 116     } scache;
 117
 118     pa_time_event *auth_timeout_event;
 119 } connection;
 120
 121 PA_DECLARE_CLASS(connection);
 122 #define CONNECTION(o) (connection_cast(o))
 123 static PA_DEFINE_CHECK_TYPE(connection, pa_msgobject);
 124
 125 struct pa_protocol_esound {
 126     pa_module *module;
 127     pa_core *core;
 128     pa_bool_t public;
 129     pa_socket_server *server;
 130     pa_idxset *connections;
 131
 132     char *sink_name, *source_name;
 133     unsigned n_player;
 134     uint8_t esd_key[ESD_KEY_LEN];
 135     pa_ip_acl *auth_ip_acl;
 136 };
 137
 138 enum {
 139     SINK_INPUT_MESSAGE_POST_DATA = PA_SINK_INPUT_MESSAGE_MAX, /* data from main loop to sink input */
 140     SINK_INPUT_MESSAGE_DISABLE_PREBUF
 141 };
 142
 143 enum {
 144     CONNECTION_MESSAGE_REQUEST_DATA,
 145     CONNECTION_MESSAGE_POST_DATA,
 146     CONNECTION_MESSAGE_UNLINK_CONNECTION
 147 };
 148
 149 typedef struct proto_handler {
 150     size_t data_length;
 151     int (*proc)(connection *c, esd_proto_t request, const void *data, size_t length);
 152     const char *description;
 153 } esd_proto_handler_info_t;
 154
 155 static int sink_input_pop_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk);
 156 static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes);
 157 static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes);
 158 static void sink_input_kill_cb(pa_sink_input *i);
 159 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
 160 static pa_usec_t source_output_get_latency_cb(pa_source_output *o);
 161
 162 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk);
 163 static void source_output_kill_cb(pa_source_output *o);
 164
 165 static int esd_proto_connect(connection *c, esd_proto_t request, const void *data, size_t length);
 166 static int esd_proto_stream_play(connection *c, esd_proto_t request, const void *data, size_t length);
 167 static int esd_proto_stream_record(connection *c, esd_proto_t request, const void *data, size_t length);
 168 static int esd_proto_get_latency(connection *c, esd_proto_t request, const void *data, size_t length);
 169 static int esd_proto_server_info(connection *c, esd_proto_t request, const void *data, size_t length);
 170 static int esd_proto_all_info(connection *c, esd_proto_t request, const void *data, size_t length);
 171 static int esd_proto_stream_pan(connection *c, esd_proto_t request, const void *data, size_t length);
 172 static int esd_proto_sample_cache(connection *c, esd_proto_t request, const void *data, size_t length);
 173 static int esd_proto_sample_free_or_play(connection *c, esd_proto_t request, const void *data, size_t length);
 174 static int esd_proto_sample_get_id(connection *c, esd_proto_t request, const void *data, size_t length);
 175 static int esd_proto_standby_or_resume(connection *c, esd_proto_t request, const void *data, size_t length);
 176
 177 /* the big map of protocol handler info */
 178 static struct proto_handler proto_map[ESD_PROTO_MAX] = {
 179     { ESD_KEY_LEN + sizeof(int),      esd_proto_connect, "connect" },
 180     { ESD_KEY_LEN + sizeof(int),      NULL, "lock" },
 181     { ESD_KEY_LEN + sizeof(int),      NULL, "unlock" },
 182
 183     { ESD_NAME_MAX + 2 * sizeof(int), esd_proto_stream_play, "stream play" },
 184     { ESD_NAME_MAX + 2 * sizeof(int), esd_proto_stream_record, "stream rec" },
 185     { ESD_NAME_MAX + 2 * sizeof(int), esd_proto_stream_record, "stream mon" },
 186
 187     { ESD_NAME_MAX + 3 * sizeof(int), esd_proto_sample_cache, "sample cache" },                      /* 6 */
 188     { sizeof(int),                    esd_proto_sample_free_or_play, "sample free" },
 189     { sizeof(int),                    esd_proto_sample_free_or_play, "sample play" },                /* 8 */
 190     { sizeof(int),                    NULL, "sample loop" },
 191     { sizeof(int),                    NULL, "sample stop" },
 192     { -1,                             NULL, "TODO: sample kill" },
 193
 194     { ESD_KEY_LEN + sizeof(int),      esd_proto_standby_or_resume, "standby" },  /* NOOP! */
 195     { ESD_KEY_LEN + sizeof(int),      esd_proto_standby_or_resume, "resume" },   /* NOOP! */         /* 13 */
 196
 197     { ESD_NAME_MAX,                   esd_proto_sample_get_id, "sample getid" },                     /* 14 */
 198     { ESD_NAME_MAX + 2 * sizeof(int), NULL, "stream filter" },
 199
 200     { sizeof(int),                    esd_proto_server_info, "server info" },
 201     { sizeof(int),                    esd_proto_all_info, "all info" },
 202     { -1,                             NULL, "TODO: subscribe" },
 203     { -1,                             NULL, "TODO: unsubscribe" },
 204
 205     { 3 * sizeof(int),                esd_proto_stream_pan, "stream pan"},
 206     { 3 * sizeof(int),                NULL, "sample pan" },
 207
 208     { sizeof(int),                    NULL, "standby mode" },
 209     { 0,                              esd_proto_get_latency, "get latency" }
 210 };
 211
 212 static void connection_unlink(connection *c) {
 213     pa_assert(c);
 214
 215     if (!c->protocol)
 216         return;
 217
 218     if (c->sink_input) {
 219         pa_sink_input_unlink(c->sink_input);
 220         pa_sink_input_unref(c->sink_input);
 221         c->sink_input = NULL;
 222     }
 223
 224     if (c->source_output) {
 225         pa_source_output_unlink(c->source_output);
 226         pa_source_output_unref(c->source_output);
 227         c->source_output = NULL;
 228     }
 229
 230     if (c->client) {
 231         pa_client_free(c->client);
 232         c->client = NULL;
 233     }
 234
 235     if (c->state == ESD_STREAMING_DATA)
 236         c->protocol->n_player--;
 237
 238     if (c->io) {
 239         pa_iochannel_free(c->io);
 240         c->io = NULL;
 241     }
 242
 243     if (c->defer_event) {
 244         c->protocol->core->mainloop->defer_free(c->defer_event);
 245         c->defer_event = NULL;
 246     }
 247
 248     if (c->auth_timeout_event) {
 249         c->protocol->core->mainloop->time_free(c->auth_timeout_event);
 250         c->auth_timeout_event = NULL;
 251     }
 252
 253     pa_assert_se(pa_idxset_remove_by_data(c->protocol->connections, c, NULL) == c);
 254     c->protocol = NULL;
 255     connection_unref(c);
 256 }
 257
 258 static void connection_free(pa_object *obj) {
 259     connection *c = CONNECTION(obj);
 260     pa_assert(c);
 261
 262     if (c->input_memblockq)
 263         pa_memblockq_free(c->input_memblockq);
 264     if (c->output_memblockq)
 265         pa_memblockq_free(c->output_memblockq);
 266
 267     if (c->playback.current_memblock)
 268         pa_memblock_unref(c->playback.current_memblock);
 269
 270     pa_xfree(c->read_data);
 271     pa_xfree(c->write_data);
 272
 273     if (c->scache.memchunk.memblock)
 274         pa_memblock_unref(c->scache.memchunk.memblock);
 275     pa_xfree(c->scache.name);
 276
 277     pa_xfree(c->original_name);
 278     pa_xfree(c);
 279 }
 280
 281 static void connection_write_prepare(connection *c, size_t length) {
 282     size_t t;
 283     pa_assert(c);
 284
 285     t = c->write_data_length+length;
 286
 287     if (c->write_data_alloc < t)
 288         c->write_data = pa_xrealloc(c->write_data, c->write_data_alloc = t);
 289
 290     pa_assert(c->write_data);
 291 }
 292
 293 static void connection_write(connection *c, const void *data, size_t length) {
 294     size_t i;
 295     pa_assert(c);
 296
 297     c->protocol->core->mainloop->defer_enable(c->defer_event, 1);
 298
 299     connection_write_prepare(c, length);
 300
 301     pa_assert(c->write_data);
 302
 303     i = c->write_data_length;
 304     c->write_data_length += length;
 305
 306     memcpy((uint8_t*) c->write_data + i, data, length);
 307 }
 308
 309 static void format_esd2native(int format, pa_bool_t swap_bytes, pa_sample_spec *ss) {
 310     pa_assert(ss);
 311
 312     ss->channels = ((format & ESD_MASK_CHAN) == ESD_STEREO) ? 2 : 1;
 313     if ((format & ESD_MASK_BITS) == ESD_BITS16)
 314         ss->format = swap_bytes ? PA_SAMPLE_S16RE : PA_SAMPLE_S16NE;
 315     else
 316         ss->format = PA_SAMPLE_U8;
 317 }
 318
 319 static int format_native2esd(pa_sample_spec *ss) {
 320     int format = 0;
 321
 322     format = (ss->format == PA_SAMPLE_U8) ? ESD_BITS8 : ESD_BITS16;
 323     format |= (ss->channels >= 2) ? ESD_STEREO : ESD_MONO;
 324
 325     return format;
 326 }
 327
 328 #define CHECK_VALIDITY(expression, ...) do { \
 329     if (!(expression)) { \
 330         pa_log_warn(__FILE__ ": " __VA_ARGS__); \
 331         return -1; \
 332     } \
 333 } while(0);
 334
 335 /*** esound commands ***/
 336
 337 static int esd_proto_connect(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 338     uint32_t ekey;
 339     int ok;
 340
 341     connection_assert_ref(c);
 342     pa_assert(data);
 343     pa_assert(length == (ESD_KEY_LEN + sizeof(uint32_t)));
 344
 345     if (!c->authorized) {
 346         if (memcmp(data, c->protocol->esd_key, ESD_KEY_LEN) != 0) {
 347             pa_log("kicked client with invalid authorization key.");
 348             return -1;
 349         }
 350
 351         c->authorized = TRUE;
 352         if (c->auth_timeout_event) {
 353             c->protocol->core->mainloop->time_free(c->auth_timeout_event);
 354             c->auth_timeout_event = NULL;
 355         }
 356     }
 357
 358     data = (const char*)data + ESD_KEY_LEN;
 359
 360     memcpy(&ekey, data, sizeof(uint32_t));
 361     if (ekey == ESD_ENDIAN_KEY)
 362         c->swap_byte_order = FALSE;
 363     else if (ekey == ESD_SWAP_ENDIAN_KEY)
 364         c->swap_byte_order = TRUE;
 365     else {
 366         pa_log_warn("Client sent invalid endian key");
 367         return -1;
 368     }
 369
 370     ok = 1;
 371     connection_write(c, &ok, sizeof(int));
 372     return 0;
 373 }
 374
 375 static int esd_proto_stream_play(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 376     char name[ESD_NAME_MAX], *utf8_name;
 377     int32_t format, rate;
 378     pa_sample_spec ss;
 379     size_t l;
 380     pa_sink *sink = NULL;
 381     pa_sink_input_new_data sdata;
 382
 383     connection_assert_ref(c);
 384     pa_assert(data);
 385     pa_assert(length == (sizeof(int32_t)*2+ESD_NAME_MAX));
 386
 387     memcpy(&format, data, sizeof(int32_t));
 388     format = PA_MAYBE_INT32_SWAP(c->swap_byte_order, format);
 389     data = (const char*) data + sizeof(int32_t);
 390
 391     memcpy(&rate, data, sizeof(int32_t));
 392     rate = PA_MAYBE_INT32_SWAP(c->swap_byte_order, rate);
 393     data = (const char*) data + sizeof(int32_t);
 394
 395     ss.rate = rate;
 396     format_esd2native(format, c->swap_byte_order, &ss);
 397
 398     CHECK_VALIDITY(pa_sample_spec_valid(&ss), "Invalid sample specification");
 399
 400     if (c->protocol->sink_name) {
 401         sink = pa_namereg_get(c->protocol->core, c->protocol->sink_name, PA_NAMEREG_SINK, 1);
 402         CHECK_VALIDITY(sink, "No such sink: %s", c->protocol->sink_name);
 403     }
 404
 405     pa_strlcpy(name, data, sizeof(name));
 406
 407     utf8_name = pa_utf8_filter(name);
 408     pa_client_set_name(c->client, utf8_name);
 409     pa_xfree(utf8_name);
 410
 411     c->original_name = pa_xstrdup(name);
 412
 413     pa_assert(!c->sink_input && !c->input_memblockq);
 414
 415     pa_sink_input_new_data_init(&sdata);
 416     sdata.driver = __FILE__;
 417     sdata.module = c->protocol->module;
 418     sdata.client = c->client;
 419     sdata.sink = sink;
 420     pa_proplist_update(sdata.proplist, PA_UPDATE_MERGE, c->client->proplist);
 421     pa_sink_input_new_data_set_sample_spec(&sdata, &ss);
 422
 423     c->sink_input = pa_sink_input_new(c->protocol->core, &sdata, 0);
 424     pa_sink_input_new_data_done(&sdata);
 425
 426     CHECK_VALIDITY(c->sink_input, "Failed to create sink input.");
 427
 428     l = (size_t) (pa_bytes_per_second(&ss)*PLAYBACK_BUFFER_SECONDS);
 429     c->input_memblockq = pa_memblockq_new(
 430             0,
 431             l,
 432             l,
 433             pa_frame_size(&ss),
 434             (size_t) -1,
 435             l/PLAYBACK_BUFFER_FRAGMENTS,
 436             0,
 437             NULL);
 438     pa_iochannel_socket_set_rcvbuf(c->io, l);
 439
 440     c->sink_input->parent.process_msg = sink_input_process_msg;
 441     c->sink_input->pop = sink_input_pop_cb;
 442     c->sink_input->process_rewind = sink_input_process_rewind_cb;
 443     c->sink_input->update_max_rewind = sink_input_update_max_rewind_cb;
 444     c->sink_input->kill = sink_input_kill_cb;
 445     c->sink_input->userdata = c;
 446
 447     pa_sink_input_set_requested_latency(c->sink_input, DEFAULT_SINK_LATENCY);
 448
 449     c->state = ESD_STREAMING_DATA;
 450
 451     c->protocol->n_player++;
 452
 453     pa_atomic_store(&c->playback.missing, pa_memblockq_missing(c->input_memblockq));
 454
 455     pa_sink_input_put(c->sink_input);
 456
 457     return 0;
 458 }
 459
 460 static int esd_proto_stream_record(connection *c, esd_proto_t request, const void *data, size_t length) {
 461     char name[ESD_NAME_MAX], *utf8_name;
 462     int32_t format, rate;
 463     pa_source *source = NULL;
 464     pa_sample_spec ss;
 465     size_t l;
 466     pa_source_output_new_data sdata;
 467
 468     connection_assert_ref(c);
 469     pa_assert(data);
 470     pa_assert(length == (sizeof(int32_t)*2+ESD_NAME_MAX));
 471
 472     memcpy(&format, data, sizeof(int32_t));
 473     format = PA_MAYBE_INT32_SWAP(c->swap_byte_order, format);
 474     data = (const char*) data + sizeof(int32_t);
 475
 476     memcpy(&rate, data, sizeof(int32_t));
 477     rate = PA_MAYBE_INT32_SWAP(c->swap_byte_order, rate);
 478     data = (const char*) data + sizeof(int32_t);
 479
 480     ss.rate = rate;
 481     format_esd2native(format, c->swap_byte_order, &ss);
 482
 483     CHECK_VALIDITY(pa_sample_spec_valid(&ss), "Invalid sample specification.");
 484
 485     if (request == ESD_PROTO_STREAM_MON) {
 486         pa_sink* sink;
 487
 488         if (!(sink = pa_namereg_get(c->protocol->core, c->protocol->sink_name, PA_NAMEREG_SINK, 1))) {
 489             pa_log("no such sink.");
 490             return -1;
 491         }
 492
 493         if (!(source = sink->monitor_source)) {
 494             pa_log("no such monitor source.");
 495             return -1;
 496         }
 497     } else {
 498         pa_assert(request == ESD_PROTO_STREAM_REC);
 499
 500         if (c->protocol->source_name) {
 501             if (!(source = pa_namereg_get(c->protocol->core, c->protocol->source_name, PA_NAMEREG_SOURCE, 1))) {
 502                 pa_log("no such source.");
 503                 return -1;
 504             }
 505         }
 506     }
 507
 508     pa_strlcpy(name, data, sizeof(name));
 509
 510     utf8_name = pa_utf8_filter(name);
 511     pa_client_set_name(c->client, utf8_name);
 512     pa_xfree(utf8_name);
 513
 514     c->original_name = pa_xstrdup(name);
 515
 516     pa_assert(!c->output_memblockq && !c->source_output);
 517
 518     pa_source_output_new_data_init(&sdata);
 519     sdata.driver = __FILE__;
 520     sdata.module = c->protocol->module;
 521     sdata.client = c->client;
 522     sdata.source = source;
 523     pa_proplist_update(sdata.proplist, PA_UPDATE_MERGE, c->client->proplist);
 524     pa_source_output_new_data_set_sample_spec(&sdata, &ss);
 525
 526     c->source_output = pa_source_output_new(c->protocol->core, &sdata, 0);
 527     pa_source_output_new_data_done(&sdata);
 528
 529     CHECK_VALIDITY(c->source_output, "Failed to create source output.");
 530
 531     l = (size_t) (pa_bytes_per_second(&ss)*RECORD_BUFFER_SECONDS);
 532     c->output_memblockq = pa_memblockq_new(
 533             0,
 534             l,
 535             l,
 536             pa_frame_size(&ss),
 537             1,
 538             0,
 539             0,
 540             NULL);
 541     pa_iochannel_socket_set_sndbuf(c->io, l);
 542
 543     c->source_output->push = source_output_push_cb;
 544     c->source_output->kill = source_output_kill_cb;
 545     c->source_output->get_latency = source_output_get_latency_cb;
 546     c->source_output->userdata = c;
 547
 548     pa_source_output_set_requested_latency(c->source_output, DEFAULT_SOURCE_LATENCY);
 549
 550     c->state = ESD_STREAMING_DATA;
 551
 552     c->protocol->n_player++;
 553
 554     pa_source_output_put(c->source_output);
 555
 556     return 0;
 557 }
 558
 559 static int esd_proto_get_latency(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 560     pa_sink *sink;
 561     int32_t latency;
 562
 563     connection_ref(c);
 564     pa_assert(!data);
 565     pa_assert(length == 0);
 566
 567     if (!(sink = pa_namereg_get(c->protocol->core, c->protocol->sink_name, PA_NAMEREG_SINK, 1)))
 568         latency = 0;
 569     else {
 570         double usec = pa_sink_get_latency(sink);
 571         latency = (int) ((usec*44100)/1000000);
 572     }
 573
 574     latency = PA_MAYBE_INT32_SWAP(c->swap_byte_order, latency);
 575     connection_write(c, &latency, sizeof(int32_t));
 576     return 0;
 577 }
 578
 579 static int esd_proto_server_info(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 580     int32_t rate = 44100, format = ESD_STEREO|ESD_BITS16;
 581     int32_t response;
 582     pa_sink *sink;
 583
 584     connection_ref(c);
 585     pa_assert(data);
 586     pa_assert(length == sizeof(int32_t));
 587
 588     if ((sink = pa_namereg_get(c->protocol->core, c->protocol->sink_name, PA_NAMEREG_SINK, 1))) {
 589         rate = sink->sample_spec.rate;
 590         format = format_native2esd(&sink->sample_spec);
 591     }
 592
 593     connection_write_prepare(c, sizeof(int32_t) * 3);
 594
 595     response = 0;
 596     connection_write(c, &response, sizeof(int32_t));
 597     rate = PA_MAYBE_INT32_SWAP(c->swap_byte_order, rate);
 598     connection_write(c, &rate, sizeof(int32_t));
 599     format = PA_MAYBE_INT32_SWAP(c->swap_byte_order, format);
 600     connection_write(c, &format, sizeof(int32_t));
 601
 602     return 0;
 603 }
 604
 605 static int esd_proto_all_info(connection *c, esd_proto_t request, const void *data, size_t length) {
 606     size_t t, k, s;
 607     connection *conn;
 608     uint32_t idx = PA_IDXSET_INVALID;
 609     unsigned nsamples;
 610     char terminator[sizeof(int32_t)*6+ESD_NAME_MAX];
 611
 612     connection_ref(c);
 613     pa_assert(data);
 614     pa_assert(length == sizeof(int32_t));
 615
 616     if (esd_proto_server_info(c, request, data, length) < 0)
 617         return -1;
 618
 619     k = sizeof(int32_t)*5+ESD_NAME_MAX;
 620     s = sizeof(int32_t)*6+ESD_NAME_MAX;
 621     nsamples = c->protocol->core->scache ? pa_idxset_size(c->protocol->core->scache) : 0;
 622     t = s*(nsamples+1) + k*(c->protocol->n_player+1);
 623
 624     connection_write_prepare(c, t);
 625
 626     memset(terminator, 0, sizeof(terminator));
 627
 628     for (conn = pa_idxset_first(c->protocol->connections, &idx); conn; conn = pa_idxset_next(c->protocol->connections, &idx)) {
 629         int32_t id, format = ESD_BITS16 | ESD_STEREO, rate = 44100, lvolume = ESD_VOLUME_BASE, rvolume = ESD_VOLUME_BASE;
 630         char name[ESD_NAME_MAX];
 631
 632         if (conn->state != ESD_STREAMING_DATA)
 633             continue;
 634
 635         pa_assert(t >= k*2+s);
 636
 637         if (conn->sink_input) {
 638             pa_cvolume volume = *pa_sink_input_get_volume(conn->sink_input);
 639             rate = conn->sink_input->sample_spec.rate;
 640             lvolume = (volume.values[0]*ESD_VOLUME_BASE)/PA_VOLUME_NORM;
 641             rvolume = (volume.values[1]*ESD_VOLUME_BASE)/PA_VOLUME_NORM;
 642             format = format_native2esd(&conn->sink_input->sample_spec);
 643         }
 644
 645         /* id */
 646         id = PA_MAYBE_INT32_SWAP(c->swap_byte_order, (int32_t) (conn->index+1));
 647         connection_write(c, &id, sizeof(int32_t));
 648
 649         /* name */
 650         memset(name, 0, ESD_NAME_MAX); /* don't leak old data */
 651         if (conn->original_name)
 652             strncpy(name, conn->original_name, ESD_NAME_MAX);
 653         else if (conn->client && pa_proplist_gets(conn->client->proplist, PA_PROP_APPLICATION_NAME))
 654             strncpy(name, pa_proplist_gets(conn->client->proplist, PA_PROP_APPLICATION_NAME), ESD_NAME_MAX);
 655         connection_write(c, name, ESD_NAME_MAX);
 656
 657         /* rate */
 658         rate = PA_MAYBE_INT32_SWAP(c->swap_byte_order, rate);
 659         connection_write(c, &rate, sizeof(int32_t));
 660
 661         /* left */
 662         lvolume = PA_MAYBE_INT32_SWAP(c->swap_byte_order, lvolume);
 663         connection_write(c, &lvolume, sizeof(int32_t));
 664
 665         /*right*/
 666         rvolume = PA_MAYBE_INT32_SWAP(c->swap_byte_order, rvolume);
 667         connection_write(c, &rvolume, sizeof(int32_t));
 668
 669         /*format*/
 670         format = PA_MAYBE_INT32_SWAP(c->swap_byte_order, format);
 671         connection_write(c, &format, sizeof(int32_t));
 672
 673         t -= k;
 674     }
 675
 676     pa_assert(t == s*(nsamples+1)+k);
 677     t -= k;
 678
 679     connection_write(c, terminator, k);
 680
 681     if (nsamples) {
 682         pa_scache_entry *ce;
 683
 684         idx = PA_IDXSET_INVALID;
 685         for (ce = pa_idxset_first(c->protocol->core->scache, &idx); ce; ce = pa_idxset_next(c->protocol->core->scache, &idx)) {
 686             int32_t id, rate, lvolume, rvolume, format, len;
 687             char name[ESD_NAME_MAX];
 688
 689             pa_assert(t >= s*2);
 690
 691             /* id */
 692             id = PA_MAYBE_INT32_SWAP(c->swap_byte_order, (int) (ce->index+1));
 693             connection_write(c, &id, sizeof(int32_t));
 694
 695             /* name */
 696             memset(name, 0, ESD_NAME_MAX); /* don't leak old data */
 697             if (strncmp(ce->name, SCACHE_PREFIX, sizeof(SCACHE_PREFIX)-1) == 0)
 698                 strncpy(name, ce->name+sizeof(SCACHE_PREFIX)-1, ESD_NAME_MAX);
 699             else
 700                 pa_snprintf(name, ESD_NAME_MAX, "native.%s", ce->name);
 701             connection_write(c, name, ESD_NAME_MAX);
 702
 703             /* rate */
 704             rate = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, ce->sample_spec.rate);
 705             connection_write(c, &rate, sizeof(int32_t));
 706
 707             /* left */
 708             lvolume = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, (ce->volume.values[0]*ESD_VOLUME_BASE)/PA_VOLUME_NORM);
 709             connection_write(c, &lvolume, sizeof(int32_t));
 710
 711             /*right*/
 712             rvolume = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, (ce->volume.values[0]*ESD_VOLUME_BASE)/PA_VOLUME_NORM);
 713             connection_write(c, &rvolume, sizeof(int32_t));
 714
 715             /*format*/
 716             format = PA_MAYBE_INT32_SWAP(c->swap_byte_order, format_native2esd(&ce->sample_spec));
 717             connection_write(c, &format, sizeof(int32_t));
 718
 719             /*length*/
 720             len = PA_MAYBE_INT32_SWAP(c->swap_byte_order, (int) ce->memchunk.length);
 721             connection_write(c, &len, sizeof(int32_t));
 722
 723             t -= s;
 724         }
 725     }
 726
 727     pa_assert(t == s);
 728
 729     connection_write(c, terminator, s);
 730
 731     return 0;
 732 }
 733
 734 static int esd_proto_stream_pan(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 735     int32_t ok;
 736     uint32_t idx, lvolume, rvolume;
 737     connection *conn;
 738
 739     connection_assert_ref(c);
 740     pa_assert(data);
 741     pa_assert(length == sizeof(int32_t)*3);
 742
 743     memcpy(&idx, data, sizeof(uint32_t));
 744     idx = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, idx) - 1;
 745     data = (const char*)data + sizeof(uint32_t);
 746
 747     memcpy(&lvolume, data, sizeof(uint32_t));
 748     lvolume = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, lvolume);
 749     data = (const char*)data + sizeof(uint32_t);
 750
 751     memcpy(&rvolume, data, sizeof(uint32_t));
 752     rvolume = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, rvolume);
 753     data = (const char*)data + sizeof(uint32_t);
 754
 755     if ((conn = pa_idxset_get_by_index(c->protocol->connections, idx)) && conn->sink_input) {
 756         pa_cvolume volume;
 757         volume.values[0] = (lvolume*PA_VOLUME_NORM)/ESD_VOLUME_BASE;
 758         volume.values[1] = (rvolume*PA_VOLUME_NORM)/ESD_VOLUME_BASE;
 759         volume.channels = 2;
 760         pa_sink_input_set_volume(conn->sink_input, &volume);
 761         ok = 1;
 762     } else
 763         ok = 0;
 764
 765     connection_write(c, &ok, sizeof(int32_t));
 766
 767     return 0;
 768 }
 769
 770 static int esd_proto_sample_cache(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 771     pa_sample_spec ss;
 772     int32_t format, rate, sc_length;
 773     uint32_t idx;
 774     char name[ESD_NAME_MAX+sizeof(SCACHE_PREFIX)-1];
 775
 776     connection_assert_ref(c);
 777     pa_assert(data);
 778     pa_assert(length == (ESD_NAME_MAX+3*sizeof(int32_t)));
 779
 780     memcpy(&format, data, sizeof(int32_t));
 781     format = PA_MAYBE_INT32_SWAP(c->swap_byte_order, format);
 782     data = (const char*)data + sizeof(int32_t);
 783
 784     memcpy(&rate, data, sizeof(int32_t));
 785     rate = PA_MAYBE_INT32_SWAP(c->swap_byte_order, rate);
 786     data = (const char*)data + sizeof(int32_t);
 787
 788     ss.rate = rate;
 789     format_esd2native(format, c->swap_byte_order, &ss);
 790
 791     CHECK_VALIDITY(pa_sample_spec_valid(&ss), "Invalid sample specification.");
 792
 793     memcpy(&sc_length, data, sizeof(int32_t));
 794     sc_length = PA_MAYBE_INT32_SWAP(c->swap_byte_order, sc_length);
 795     data = (const char*)data + sizeof(int32_t);
 796
 797     CHECK_VALIDITY(sc_length <= MAX_CACHE_SAMPLE_SIZE, "Sample too large (%d bytes).", (int)sc_length);
 798
 799     strcpy(name, SCACHE_PREFIX);
 800     pa_strlcpy(name+sizeof(SCACHE_PREFIX)-1, data, ESD_NAME_MAX);
 801
 802     CHECK_VALIDITY(pa_utf8_valid(name), "Invalid UTF8 in sample name.");
 803
 804     pa_assert(!c->scache.memchunk.memblock);
 805     c->scache.memchunk.memblock = pa_memblock_new(c->protocol->core->mempool, sc_length);
 806     c->scache.memchunk.index = 0;
 807     c->scache.memchunk.length = sc_length;
 808     c->scache.sample_spec = ss;
 809     pa_assert(!c->scache.name);
 810     c->scache.name = pa_xstrdup(name);
 811
 812     c->state = ESD_CACHING_SAMPLE;
 813
 814     pa_scache_add_item(c->protocol->core, c->scache.name, NULL, NULL, NULL, c->client->proplist, &idx);
 815
 816     idx += 1;
 817     connection_write(c, &idx, sizeof(uint32_t));
 818
 819     return 0;
 820 }
 821
 822 static int esd_proto_sample_get_id(connection *c, PA_GCC_UNUSED esd_proto_t request, const void *data, size_t length) {
 823     int32_t ok;
 824     uint32_t idx;
 825     char name[ESD_NAME_MAX+sizeof(SCACHE_PREFIX)-1];
 826
 827     connection_assert_ref(c);
 828     pa_assert(data);
 829     pa_assert(length == ESD_NAME_MAX);
 830
 831     strcpy(name, SCACHE_PREFIX);
 832     pa_strlcpy(name+sizeof(SCACHE_PREFIX)-1, data, ESD_NAME_MAX);
 833
 834     CHECK_VALIDITY(pa_utf8_valid(name), "Invalid UTF8 in sample name.");
 835
 836     ok = -1;
 837     if ((idx = pa_scache_get_id_by_name(c->protocol->core, name)) != PA_IDXSET_INVALID)
 838         ok = idx + 1;
 839
 840     connection_write(c, &ok, sizeof(int32_t));
 841
 842     return 0;
 843 }
 844
 845 static int esd_proto_sample_free_or_play(connection *c, esd_proto_t request, const void *data, size_t length) {
 846     int32_t ok;
 847     const char *name;
 848     uint32_t idx;
 849
 850     connection_assert_ref(c);
 851     pa_assert(data);
 852     pa_assert(length == sizeof(int32_t));
 853
 854     memcpy(&idx, data, sizeof(uint32_t));
 855     idx = PA_MAYBE_UINT32_SWAP(c->swap_byte_order, idx) - 1;
 856
 857     ok = 0;
 858
 859     if ((name = pa_scache_get_name_by_id(c->protocol->core, idx))) {
 860         if (request == ESD_PROTO_SAMPLE_PLAY) {
 861             pa_sink *sink;
 862
 863             if ((sink = pa_namereg_get(c->protocol->core, c->protocol->sink_name, PA_NAMEREG_SINK, 1)))
 864                 if (pa_scache_play_item(c->protocol->core, name, sink, PA_VOLUME_NORM, c->client->proplist, NULL) >= 0)
 865                     ok = idx + 1;
 866         } else {
 867             pa_assert(request == ESD_PROTO_SAMPLE_FREE);
 868
 869             if (pa_scache_remove_item(c->protocol->core, name) >= 0)
 870                 ok = idx + 1;
 871         }
 872     }
 873
 874     connection_write(c, &ok, sizeof(int32_t));
 875
 876     return 0;
 877 }
 878
 879 static int esd_proto_standby_or_resume(connection *c, PA_GCC_UNUSED esd_proto_t request, PA_GCC_UNUSED const void *data, PA_GCC_UNUSED size_t length) {
 880     int32_t ok;
 881
 882     connection_assert_ref(c);
 883
 884     connection_write_prepare(c, sizeof(int32_t) * 2);
 885
 886     ok = 1;
 887     connection_write(c, &ok, sizeof(int32_t));
 888     connection_write(c, &ok, sizeof(int32_t));
 889
 890     return 0;
 891 }
 892
 893 /*** client callbacks ***/
 894
 895 static void client_kill_cb(pa_client *c) {
 896     pa_assert(c);
 897
 898     connection_unlink(CONNECTION(c->userdata));
 899 }
 900
 901 /*** pa_iochannel callbacks ***/
 902
 903 static int do_read(connection *c) {
 904     connection_assert_ref(c);
 905
 906 /*     pa_log("READ"); */
 907
 908     if (c->state == ESD_NEXT_REQUEST) {
 909         ssize_t r;
 910         pa_assert(c->read_data_length < sizeof(c->request));
 911
 912         if ((r = pa_iochannel_read(c->io, ((uint8_t*) &c->request) + c->read_data_length, sizeof(c->request) - c->read_data_length)) <= 0) {
 913             pa_log_debug("read(): %s", r < 0 ? pa_cstrerror(errno) : "EOF");
 914             return -1;
 915         }
 916
 917         if ((c->read_data_length+= r) >= sizeof(c->request)) {
 918             struct proto_handler *handler;
 919
 920             c->request = PA_MAYBE_INT32_SWAP(c->swap_byte_order, c->request);
 921
 922             if (c->request < ESD_PROTO_CONNECT || c->request > ESD_PROTO_MAX) {
 923                 pa_log("recieved invalid request.");
 924                 return -1;
 925             }
 926
 927             handler = proto_map+c->request;
 928
 929 /*             pa_log("executing request #%u", c->request); */
 930
 931             if (!handler->proc) {
 932                 pa_log("recieved unimplemented request #%u.", c->request);
 933                 return -1;
 934             }
 935
 936             if (handler->data_length == 0) {
 937                 c->read_data_length = 0;
 938
 939                 if (handler->proc(c, c->request, NULL, 0) < 0)
 940                     return -1;
 941
 942             } else {
 943                 if (c->read_data_alloc < handler->data_length)
 944                     c->read_data = pa_xrealloc(c->read_data, c->read_data_alloc = handler->data_length);
 945                 pa_assert(c->read_data);
 946
 947                 c->state = ESD_NEEDS_REQDATA;
 948                 c->read_data_length = 0;
 949             }
 950         }
 951
 952     } else if (c->state == ESD_NEEDS_REQDATA) {
 953         ssize_t r;
 954         struct proto_handler *handler = proto_map+c->request;
 955
 956         pa_assert(handler->proc);
 957
 958         pa_assert(c->read_data && c->read_data_length < handler->data_length);
 959
 960         if ((r = pa_iochannel_read(c->io, (uint8_t*) c->read_data + c->read_data_length, handler->data_length - c->read_data_length)) <= 0) {
 961             if (r < 0 && (errno == EINTR || errno == EAGAIN))
 962                 return 0;
 963
 964             pa_log_debug("read(): %s", r < 0 ? pa_cstrerror(errno) : "EOF");
 965             return -1;
 966         }
 967
 968         if ((c->read_data_length += r) >= handler->data_length) {
 969             size_t l = c->read_data_length;
 970             pa_assert(handler->proc);
 971
 972             c->state = ESD_NEXT_REQUEST;
 973             c->read_data_length = 0;
 974
 975             if (handler->proc(c, c->request, c->read_data, l) < 0)
 976                 return -1;
 977         }
 978     } else if (c->state == ESD_CACHING_SAMPLE) {
 979         ssize_t r;
 980         void *p;
 981
 982         pa_assert(c->scache.memchunk.memblock);
 983         pa_assert(c->scache.name);
 984         pa_assert(c->scache.memchunk.index < c->scache.memchunk.length);
 985
 986         p = pa_memblock_acquire(c->scache.memchunk.memblock);
 987         r = pa_iochannel_read(c->io, (uint8_t*) p+c->scache.memchunk.index, c->scache.memchunk.length-c->scache.memchunk.index);
 988         pa_memblock_release(c->scache.memchunk.memblock);
 989
 990         if (r <= 0) {
 991             if (r < 0 && (errno == EINTR || errno == EAGAIN))
 992                 return 0;
 993
 994             pa_log_debug("read(): %s", r < 0 ? pa_cstrerror(errno) : "EOF");
 995             return -1;
 996         }
 997
 998         c->scache.memchunk.index += r;
 999         pa_assert(c->scache.memchunk.index <= c->scache.memchunk.length);
1000
1001         if (c->scache.memchunk.index == c->scache.memchunk.length) {
1002             uint32_t idx;
1003
1004             c->scache.memchunk.index = 0;
1005             pa_scache_add_item(c->protocol->core, c->scache.name, &c->scache.sample_spec, NULL, &c->scache.memchunk, c->client->proplist, &idx);
1006
1007             pa_memblock_unref(c->scache.memchunk.memblock);
1008             c->scache.memchunk.memblock = NULL;
1009             c->scache.memchunk.index = c->scache.memchunk.length = 0;
1010
1011             pa_xfree(c->scache.name);
1012             c->scache.name = NULL;
1013
1014             c->state = ESD_NEXT_REQUEST;
1015
1016             idx += 1;
1017             connection_write(c, &idx, sizeof(uint32_t));
1018         }
1019
1020     } else if (c->state == ESD_STREAMING_DATA && c->sink_input) {
1021         pa_memchunk chunk;
1022         ssize_t r;
1023         size_t l;
1024         void *p;
1025         size_t space;
1026
1027         pa_assert(c->input_memblockq);
1028
1029 /*         pa_log("STREAMING_DATA"); */
1030
1031         if (!(l = pa_atomic_load(&c->playback.missing)))
1032             return 0;
1033
1034         if (c->playback.current_memblock) {
1035
1036             space = pa_memblock_get_length(c->playback.current_memblock) - c->playback.memblock_index;
1037
1038             if (space <= 0) {
1039                 pa_memblock_unref(c->playback.current_memblock);
1040                 c->playback.current_memblock = NULL;
1041             }
1042         }
1043
1044         if (!c->playback.current_memblock) {
1045             pa_assert_se(c->playback.current_memblock = pa_memblock_new(c->protocol->core->mempool, 0));
1046             c->playback.memblock_index = 0;
1047
1048             space = pa_memblock_get_length(c->playback.current_memblock);
1049         }
1050
1051         if (l > space)
1052             l = space;
1053
1054         p = pa_memblock_acquire(c->playback.current_memblock);
1055         r = pa_iochannel_read(c->io, (uint8_t*) p+c->playback.memblock_index, l);
1056         pa_memblock_release(c->playback.current_memblock);
1057
1058         if (r <= 0) {
1059
1060             if (r < 0 && (errno == EINTR || errno == EAGAIN))
1061                 return 0;
1062
1063             pa_log_debug("read(): %s", r < 0 ? pa_cstrerror(errno) : "EOF");
1064             return -1;
1065         }
1066
1067         chunk.memblock = c->playback.current_memblock;
1068         chunk.index = c->playback.memblock_index;
1069         chunk.length = r;
1070
1071         c->playback.memblock_index += r;
1072
1073         pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, &chunk, NULL);
1074         pa_atomic_sub(&c->playback.missing, r);
1075     }
1076
1077     return 0;
1078 }
1079
1080 static int do_write(connection *c) {
1081     connection_assert_ref(c);
1082
1083 /*     pa_log("WRITE"); */
1084
1085     if (c->write_data_length) {
1086         ssize_t r;
1087
1088         pa_assert(c->write_data_index < c->write_data_length);
1089         if ((r = pa_iochannel_write(c->io, (uint8_t*) c->write_data+c->write_data_index, c->write_data_length-c->write_data_index)) < 0) {
1090
1091             if (r < 0 && (errno == EINTR || errno == EAGAIN))
1092                 return 0;
1093
1094             pa_log("write(): %s", pa_cstrerror(errno));
1095             return -1;
1096         }
1097
1098         if ((c->write_data_index +=r) >= c->write_data_length)
1099             c->write_data_length = c->write_data_index = 0;
1100
1101     } else if (c->state == ESD_STREAMING_DATA && c->source_output) {
1102         pa_memchunk chunk;
1103         ssize_t r;
1104         void *p;
1105
1106         if (pa_memblockq_peek(c->output_memblockq, &chunk) < 0)
1107             return 0;
1108
1109         pa_assert(chunk.memblock);
1110         pa_assert(chunk.length);
1111
1112         p = pa_memblock_acquire(chunk.memblock);
1113         r = pa_iochannel_write(c->io, (uint8_t*) p+chunk.index, chunk.length);
1114         pa_memblock_release(chunk.memblock);
1115
1116         pa_memblock_unref(chunk.memblock);
1117
1118         if (r < 0) {
1119
1120             if (r < 0 && (errno == EINTR || errno == EAGAIN))
1121                 return 0;
1122
1123             pa_log("write(): %s", pa_cstrerror(errno));
1124             return -1;
1125         }
1126
1127         pa_memblockq_drop(c->output_memblockq, r);
1128     }
1129
1130     return 0;
1131 }
1132
1133 static void do_work(connection *c) {
1134     connection_assert_ref(c);
1135
1136     c->protocol->core->mainloop->defer_enable(c->defer_event, 0);
1137
1138     if (c->dead)
1139         return;
1140
1141     if (pa_iochannel_is_readable(c->io))
1142         if (do_read(c) < 0)
1143             goto fail;
1144
1145     if (c->state == ESD_STREAMING_DATA && !c->sink_input && pa_iochannel_is_hungup(c->io))
1146         /* In case we are in capture mode we will never call read()
1147          * on the socket, hence we need to detect the hangup manually
1148          * here, instead of simply waiting for read() to return 0. */
1149         goto fail;
1150
1151     if (pa_iochannel_is_writable(c->io))
1152         if (do_write(c) < 0)
1153             goto fail;
1154
1155     return;
1156
1157 fail:
1158
1159     if (c->state == ESD_STREAMING_DATA && c->sink_input) {
1160         c->dead = TRUE;
1161
1162         pa_iochannel_free(c->io);
1163         c->io = NULL;
1164
1165         pa_asyncmsgq_post(c->sink_input->sink->asyncmsgq, PA_MSGOBJECT(c->sink_input), SINK_INPUT_MESSAGE_DISABLE_PREBUF, NULL, 0, NULL, NULL);
1166     } else
1167         connection_unlink(c);
1168 }
1169
1170 static void io_callback(pa_iochannel*io, void *userdata) {
1171     connection *c = CONNECTION(userdata);
1172
1173     connection_assert_ref(c);
1174     pa_assert(io);
1175
1176     do_work(c);
1177 }
1178
1179 static void defer_callback(pa_mainloop_api*a, pa_defer_event *e, void *userdata) {
1180     connection *c = CONNECTION(userdata);
1181
1182     connection_assert_ref(c);
1183     pa_assert(e);
1184
1185     do_work(c);
1186 }
1187
1188 static int connection_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
1189     connection *c = CONNECTION(o);
1190     connection_assert_ref(c);
1191
1192     switch (code) {
1193         case CONNECTION_MESSAGE_REQUEST_DATA:
1194             do_work(c);
1195             break;
1196
1197         case CONNECTION_MESSAGE_POST_DATA:
1198 /*             pa_log("got data %u", chunk->length); */
1199             pa_memblockq_push_align(c->output_memblockq, chunk);
1200             do_work(c);
1201             break;
1202
1203         case CONNECTION_MESSAGE_UNLINK_CONNECTION:
1204             connection_unlink(c);
1205             break;
1206     }
1207
1208     return 0;
1209 }
1210
1211 /*** sink_input callbacks ***/
1212
1213 /* Called from thread context */
1214 static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
1215     pa_sink_input *i = PA_SINK_INPUT(o);
1216     connection*c;
1217
1218     pa_sink_input_assert_ref(i);
1219     c = CONNECTION(i->userdata);
1220     connection_assert_ref(c);
1221
1222     switch (code) {
1223
1224         case SINK_INPUT_MESSAGE_POST_DATA: {
1225             pa_assert(chunk);
1226
1227             /* New data from the main loop */
1228             pa_memblockq_push_align(c->input_memblockq, chunk);
1229
1230             if (pa_memblockq_is_readable(c->input_memblockq) && c->playback.underrun) {
1231                 pa_log_debug("Requesting rewind due to end of underrun.");
1232                 pa_sink_input_request_rewind(c->sink_input, 0, FALSE, TRUE);
1233             }
1234
1235 /*             pa_log("got data, %u", pa_memblockq_get_length(c->input_memblockq)); */
1236
1237             return 0;
1238         }
1239
1240         case SINK_INPUT_MESSAGE_DISABLE_PREBUF:
1241             pa_memblockq_prebuf_disable(c->input_memblockq);
1242             return 0;
1243
1244         case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
1245             pa_usec_t *r = userdata;
1246
1247             *r = pa_bytes_to_usec(pa_memblockq_get_length(c->input_memblockq), &c->sink_input->sample_spec);
1248
1249             /* Fall through, the default handler will add in the extra
1250              * latency added by the resampler */
1251         }
1252
1253         default:
1254             return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
1255     }
1256 }
1257
1258 /* Called from thread context */
1259 static int sink_input_pop_cb(pa_sink_input *i, size_t length, pa_memchunk *chunk) {
1260     connection*c;
1261
1262     pa_sink_input_assert_ref(i);
1263     c = CONNECTION(i->userdata);
1264     connection_assert_ref(c);
1265     pa_assert(chunk);
1266
1267     if (pa_memblockq_peek(c->input_memblockq, chunk) < 0) {
1268
1269         c->playback.underrun = TRUE;
1270
1271         if (c->dead && pa_sink_input_safe_to_remove(i))
1272             pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_UNLINK_CONNECTION, NULL, 0, NULL, NULL);
1273
1274         return -1;
1275     } else {
1276         size_t m;
1277
1278         c->playback.underrun = FALSE;
1279
1280         pa_memblockq_drop(c->input_memblockq, chunk->length);
1281         m = pa_memblockq_pop_missing(c->input_memblockq);
1282
1283         if (m > 0)
1284             if (pa_atomic_add(&c->playback.missing, m) <= 0)
1285                 pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
1286
1287         return 0;
1288     }
1289 }
1290
1291 /* Called from thread context */
1292 static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
1293     connection *c;
1294
1295     pa_sink_input_assert_ref(i);
1296     c = CONNECTION(i->userdata);
1297     connection_assert_ref(c);
1298
1299     /* If we are in an underrun, then we don't rewind */
1300     if (i->thread_info.underrun_for > 0)
1301         return;
1302
1303     pa_memblockq_rewind(c->input_memblockq, nbytes);
1304 }
1305
1306 /* Called from thread context */
1307 static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
1308     connection *c;
1309
1310     pa_sink_input_assert_ref(i);
1311     c = CONNECTION(i->userdata);
1312     connection_assert_ref(c);
1313
1314     pa_memblockq_set_maxrewind(c->input_memblockq, nbytes);
1315 }
1316
1317 static void sink_input_kill_cb(pa_sink_input *i) {
1318     pa_sink_input_assert_ref(i);
1319
1320     connection_unlink(CONNECTION(i->userdata));
1321 }
1322
1323 /*** source_output callbacks ***/
1324
1325 /* Called from thread context */
1326 static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
1327     connection *c;
1328
1329     pa_source_output_assert_ref(o);
1330     c = CONNECTION(o->userdata);
1331     pa_assert(c);
1332     pa_assert(chunk);
1333
1334     pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(c), CONNECTION_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
1335 }
1336
1337 static void source_output_kill_cb(pa_source_output *o) {
1338     pa_source_output_assert_ref(o);
1339
1340     connection_unlink(CONNECTION(o->userdata));
1341 }
1342
1343 static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
1344     connection*c;
1345
1346     pa_source_output_assert_ref(o);
1347     c = CONNECTION(o->userdata);
1348     pa_assert(c);
1349
1350     return pa_bytes_to_usec(pa_memblockq_get_length(c->output_memblockq), &c->source_output->sample_spec);
1351 }
1352
1353 /*** socket server callback ***/
1354
1355 static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *tv, void *userdata) {
1356     connection *c = CONNECTION(userdata);
1357
1358     pa_assert(m);
1359     pa_assert(tv);
1360     connection_assert_ref(c);
1361     pa_assert(c->auth_timeout_event == e);
1362
1363     if (!c->authorized)
1364         connection_unlink(c);
1365 }
1366
1367 static void on_connection(pa_socket_server*s, pa_iochannel *io, void *userdata) {
1368     connection *c;
1369     pa_protocol_esound *p = userdata;
1370     char cname[256], pname[128];
1371
1372     pa_assert(s);
1373     pa_assert(io);
1374     pa_assert(p);
1375
1376     if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
1377         pa_log("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
1378         pa_iochannel_free(io);
1379         return;
1380     }
1381
1382     c = pa_msgobject_new(connection);
1383     c->parent.parent.free = connection_free;
1384     c->parent.process_msg = connection_process_msg;
1385     c->protocol = p;
1386     c->io = io;
1387     pa_iochannel_set_callback(c->io, io_callback, c);
1388
1389     pa_iochannel_socket_peer_to_string(io, pname, sizeof(pname));
1390     pa_snprintf(cname, sizeof(cname), "EsounD client (%s)", pname);
1391     c->client = pa_client_new(p->core, __FILE__, cname);
1392     pa_proplist_sets(c->client->proplist, "esound-protocol.peer", pname);
1393     c->client->module = p->module;
1394     c->client->kill = client_kill_cb;
1395     c->client->userdata = c;
1396
1397     c->authorized = !!p->public;
1398     c->swap_byte_order = FALSE;
1399     c->dead = FALSE;
1400
1401     c->read_data_length = 0;
1402     c->read_data = pa_xmalloc(c->read_data_alloc = proto_map[ESD_PROTO_CONNECT].data_length);
1403
1404     c->write_data_length = c->write_data_index = c->write_data_alloc = 0;
1405     c->write_data = NULL;
1406
1407     c->state = ESD_NEEDS_REQDATA;
1408     c->request = ESD_PROTO_CONNECT;
1409
1410     c->sink_input = NULL;
1411     c->input_memblockq = NULL;
1412
1413     c->source_output = NULL;
1414     c->output_memblockq = NULL;
1415
1416     c->playback.current_memblock = NULL;
1417     c->playback.memblock_index = 0;
1418     c->playback.underrun = TRUE;
1419     pa_atomic_store(&c->playback.missing, 0);
1420
1421     pa_memchunk_reset(&c->scache.memchunk);
1422     c->scache.name = NULL;
1423
1424     c->original_name = NULL;
1425
1426     if (!c->authorized && p->auth_ip_acl && pa_ip_acl_check(p->auth_ip_acl, pa_iochannel_get_recv_fd(io)) > 0) {
1427         pa_log_info("Client authenticated by IP ACL.");
1428         c->authorized = TRUE;
1429     }
1430
1431     if (!c->authorized) {
1432         struct timeval tv;
1433         pa_gettimeofday(&tv);
1434         tv.tv_sec += AUTH_TIMEOUT;
1435         c->auth_timeout_event = p->core->mainloop->time_new(p->core->mainloop, &tv, auth_timeout, c);
1436     } else
1437         c->auth_timeout_event = NULL;
1438
1439     c->defer_event = p->core->mainloop->defer_new(p->core->mainloop, defer_callback, c);
1440     p->core->mainloop->defer_enable(c->defer_event, 0);
1441
1442     pa_idxset_put(p->connections, c, &c->index);
1443 }
1444
1445 /*** entry points ***/
1446
1447 pa_protocol_esound* pa_protocol_esound_new(pa_core*core, pa_socket_server *server, pa_module *m, pa_modargs *ma) {
1448     pa_protocol_esound *p = NULL;
1449     pa_bool_t public = FALSE;
1450     const char *acl;
1451
1452     pa_assert(core);
1453     pa_assert(server);
1454     pa_assert(m);
1455     pa_assert(ma);
1456
1457     if (pa_modargs_get_value_boolean(ma, "auth-anonymous", &public) < 0) {
1458         pa_log("auth-anonymous= expects a boolean argument.");
1459         goto fail;
1460     }
1461
1462     p = pa_xnew(pa_protocol_esound, 1);
1463
1464     if (pa_authkey_load_auto(pa_modargs_get_value(ma, "cookie", DEFAULT_COOKIE_FILE), p->esd_key, sizeof(p->esd_key)) < 0)
1465         goto fail;
1466
1467     if ((acl = pa_modargs_get_value(ma, "auth-ip-acl", NULL))) {
1468
1469         if (!(p->auth_ip_acl = pa_ip_acl_new(acl))) {
1470             pa_log("Failed to parse IP ACL '%s'", acl);
1471             goto fail;
1472         }
1473     } else
1474         p->auth_ip_acl = NULL;
1475
1476     p->core = core;
1477     p->module = m;
1478     p->public = public;
1479     p->server = pa_socket_server_ref(server);
1480     pa_socket_server_set_callback(p->server, on_connection, p);
1481     p->connections = pa_idxset_new(NULL, NULL);
1482
1483     p->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
1484     p->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));
1485     p->n_player = 0;
1486
1487     return p;
1488
1489 fail:
1490     pa_xfree(p);
1491     return NULL;
1492 }
1493
1494 void pa_protocol_esound_free(pa_protocol_esound *p) {
1495     connection *c;
1496     pa_assert(p);
1497
1498     while ((c = pa_idxset_first(p->connections, NULL)))
1499         connection_unlink(c);
1500     pa_idxset_free(p->connections, NULL, NULL);
1501
1502     if (p->server)
1503         pa_socket_server_unref(p->server);
1504
1505     if (p->auth_ip_acl)
1506         pa_ip_acl_free(p->auth_ip_acl);
1507
1508     pa_xfree(p->sink_name);
1509     pa_xfree(p->source_name);
1510
1511     pa_xfree(p);
1512 }