#include <string.h>
#include <stdio.h>
-#include <assert.h>
#include <stdlib.h>
#include <unistd.h>
#define MAX_MEMBLOCKQ_LENGTH (4*1024*1024) /* 4MB */
-struct connection;
+typedef struct connection connection;
struct pa_protocol_native;
-struct record_stream {
- struct connection *connection;
+typedef struct record_stream {
+ pa_msgobject parent;
+
+ connection *connection;
uint32_t index;
+
pa_source_output *source_output;
pa_memblockq *memblockq;
size_t fragment_size;
-};
+} record_stream;
+
+typedef struct output_stream {
+ pa_msgobject parent;
+} output_stream;
-struct playback_stream {
- int type;
- struct connection *connection;
+typedef struct playback_stream {
+ output_stream parent;
+
+ connection *connection;
uint32_t index;
+
pa_sink_input *sink_input;
pa_memblockq *memblockq;
- size_t requested_bytes;
int drain_request;
uint32_t drain_tag;
uint32_t syncid;
int underrun;
- /* Sync group members */
- PA_LLIST_FIELDS(struct playback_stream);
-};
+ pa_atomic_t missing;
+ size_t last_missing;
+} playback_stream;
-struct upload_stream {
- int type;
- struct connection *connection;
+typedef struct upload_stream {
+ output_stream parent;
+
+ connection *connection;
uint32_t index;
+
pa_memchunk memchunk;
size_t length;
char *name;
pa_sample_spec sample_spec;
pa_channel_map channel_map;
-};
-
-struct output_stream {
- int type;
-};
-
-enum {
- UPLOAD_STREAM,
- PLAYBACK_STREAM
-};
+} upload_stream;
struct connection {
+ pa_msgobject parent;
+
int authorized;
uint32_t version;
pa_protocol_native *protocol;
pa_time_event *auth_timeout_event;
};
+
+PA_DECLARE_CLASS(record_stream);
+#define RECORD_STREAM(o) (record_stream_cast(o))
+static PA_DEFINE_CHECK_TYPE(record_stream, pa_msgobject);
+
+PA_DECLARE_CLASS(output_stream);
+#define OUTPUT_STREAM(o) (output_stream_cast(o))
+static PA_DEFINE_CHECK_TYPE(output_stream, pa_msgobject);
+
+PA_DECLARE_CLASS(playback_stream);
+#define PLAYBACK_STREAM(o) (playback_stream_cast(o))
+static PA_DEFINE_CHECK_TYPE(playback_stream, output_stream);
+
+PA_DECLARE_CLASS(upload_stream);
+#define UPLOAD_STREAM(o) (upload_stream_cast(o))
+static PA_DEFINE_CHECK_TYPE(upload_stream, output_stream);
+
+PA_DECLARE_CLASS(connection);
+#define CONNECTION(o) (connection_cast(o))
+static PA_DEFINE_CHECK_TYPE(connection, pa_msgobject);
+
struct pa_protocol_native {
pa_module *module;
- int public;
pa_core *core;
+ int public;
pa_socket_server *server;
pa_idxset *connections;
uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
pa_ip_acl *auth_ip_acl;
};
+enum {
+ SINK_INPUT_MESSAGE_POST_DATA = PA_SINK_INPUT_MESSAGE_MAX, /* data from main loop to sink input */
+ SINK_INPUT_MESSAGE_DRAIN, /* disabled prebuf, get playback started. */
+ SINK_INPUT_MESSAGE_FLUSH,
+ SINK_INPUT_MESSAGE_TRIGGER,
+ SINK_INPUT_MESSAGE_SEEK,
+ SINK_INPUT_MESSAGE_PREBUF_FORCE
+};
+
+enum {
+ PLAYBACK_STREAM_MESSAGE_REQUEST_DATA, /* data requested from sink input from the main loop */
+ PLAYBACK_STREAM_MESSAGE_UNDERFLOW,
+ PLAYBACK_STREAM_MESSAGE_OVERFLOW,
+ PLAYBACK_STREAM_MESSAGE_DRAIN_ACK
+};
+
+enum {
+ RECORD_STREAM_MESSAGE_POST_DATA /* data from source output to main loop */
+};
+
static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk);
-static void sink_input_drop_cb(pa_sink_input *i, const pa_memchunk *chunk, size_t length);
+static void sink_input_drop_cb(pa_sink_input *i, size_t length);
static void sink_input_kill_cb(pa_sink_input *i);
-static pa_usec_t sink_input_get_latency_cb(pa_sink_input *i);
+static void send_memblock(connection *c);
static void request_bytes(struct playback_stream*s);
static void source_output_kill_cb(pa_source_output *o);
static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk);
static pa_usec_t source_output_get_latency_cb(pa_source_output *o);
+static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
+
static void command_exit(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_create_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_drain_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_set_volume(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_set_mute(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_cork_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
-static void command_flush_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
-static void command_trigger_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
+static void command_trigger_or_flush_or_prebuf_playback_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_set_default_sink_or_source(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_set_stream_name(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
[PA_COMMAND_SET_SOURCE_VOLUME] = command_set_volume,
[PA_COMMAND_SET_SINK_MUTE] = command_set_mute,
+ [PA_COMMAND_SET_SINK_INPUT_MUTE] = command_set_mute,
[PA_COMMAND_SET_SOURCE_MUTE] = command_set_mute,
[PA_COMMAND_CORK_PLAYBACK_STREAM] = command_cork_playback_stream,
- [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = command_flush_playback_stream,
- [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = command_trigger_or_prebuf_playback_stream,
- [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = command_trigger_or_prebuf_playback_stream,
+ [PA_COMMAND_FLUSH_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
+ [PA_COMMAND_TRIGGER_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
+ [PA_COMMAND_PREBUF_PLAYBACK_STREAM] = command_trigger_or_flush_or_prebuf_playback_stream,
[PA_COMMAND_CORK_RECORD_STREAM] = command_cork_record_stream,
[PA_COMMAND_FLUSH_RECORD_STREAM] = command_flush_record_stream,
/* structure management */
-static struct upload_stream* upload_stream_new(
- struct connection *c,
- const pa_sample_spec *ss,
- const pa_channel_map *map,
- const char *name, size_t length) {
+static void upload_stream_unlink(upload_stream *s) {
+ pa_assert(s);
+
+ if (!s->connection)
+ return;
+
+ pa_assert_se(pa_idxset_remove_by_data(s->connection->output_streams, s, NULL) == s);
+ upload_stream_unref(s);
+ s->connection = NULL;
+}
+
+static void upload_stream_free(pa_object *o) {
+ upload_stream *s = UPLOAD_STREAM(o);
+ pa_assert(s);
- struct upload_stream *s;
- assert(c && ss && name && length);
+ upload_stream_unlink(s);
- s = pa_xnew(struct upload_stream, 1);
- s->type = UPLOAD_STREAM;
+ pa_xfree(s->name);
+
+ if (s->memchunk.memblock)
+ pa_memblock_unref(s->memchunk.memblock);
+
+ pa_xfree(s);
+}
+
+static upload_stream* upload_stream_new(
+ connection *c,
+ const pa_sample_spec *ss,
+ const pa_channel_map *map,
+ const char *name, size_t length) {
+
+ upload_stream *s;
+
+ pa_assert(c);
+ pa_assert(ss);
+ pa_assert(name);
+ pa_assert(length > 0);
+
+ s = pa_msgobject_new(upload_stream);
+ c->parent.parent.free = upload_stream_free;
s->connection = c;
s->sample_spec = *ss;
s->channel_map = *map;
s->name = pa_xstrdup(name);
-
- s->memchunk.memblock = NULL;
- s->memchunk.index = 0;
- s->memchunk.length = 0;
-
+ pa_memchunk_reset(&s->memchunk);
s->length = length;
pa_idxset_put(c->output_streams, s, &s->index);
+
return s;
}
-static void upload_stream_free(struct upload_stream *o) {
- assert(o && o->connection);
+static void record_stream_unlink(record_stream *s) {
+ pa_assert(s);
- pa_idxset_remove_by_data(o->connection->output_streams, o, NULL);
+ if (!s->connection)
+ return;
- pa_xfree(o->name);
+ if (s->source_output) {
+ pa_source_output_disconnect(s->source_output);
+ pa_source_output_unref(s->source_output);
+ s->source_output = NULL;
+ }
+
+ pa_assert_se(pa_idxset_remove_by_data(s->connection->record_streams, s, NULL) == s);
+ record_stream_unref(s);
+ s->connection = NULL;
+}
+
+static void record_stream_free(pa_object *o) {
+ record_stream *s = RECORD_STREAM(o);
+ pa_assert(s);
+
+ record_stream_unlink(s);
+
+ pa_memblockq_free(s->memblockq);
+ pa_xfree(s);
+}
- if (o->memchunk.memblock)
- pa_memblock_unref(o->memchunk.memblock);
+static int record_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
+ record_stream *s = RECORD_STREAM(o);
+ record_stream_assert_ref(s);
+
+ switch (code) {
+
+ case RECORD_STREAM_MESSAGE_POST_DATA:
+
+ if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
+ pa_log_warn("Failed to push data into output queue.");
+ return -1;
+ }
+
+ if (!pa_pstream_is_pending(s->connection->pstream))
+ send_memblock(s->connection);
+
+ pa_pstream_send_memblock(s->connection->pstream, s->index, 0, PA_SEEK_RELATIVE, chunk);
+ break;
+ }
- pa_xfree(o);
+ return 0;
}
-static struct record_stream* record_stream_new(
- struct connection *c,
- pa_source *source,
- const pa_sample_spec *ss,
- const pa_channel_map *map,
- const char *name,
- size_t maxlength,
- size_t fragment_size) {
+static record_stream* record_stream_new(
+ connection *c,
+ pa_source *source,
+ const pa_sample_spec *ss,
+ const pa_channel_map *map,
+ const char *name,
+ size_t *maxlength,
+ size_t fragment_size,
+ int corked) {
- struct record_stream *s;
+ record_stream *s;
pa_source_output *source_output;
size_t base;
pa_source_output_new_data data;
- assert(c && ss && name && maxlength);
+ pa_assert(c);
+ pa_assert(ss);
+ pa_assert(name);
+ pa_assert(maxlength);
+ pa_assert(*maxlength > 0);
pa_source_output_new_data_init(&data);
+ data.module = c->protocol->module;
+ data.client = c->client;
data.source = source;
data.driver = __FILE__;
data.name = name;
+ data.corked = corked;
pa_source_output_new_data_set_sample_spec(&data, ss);
pa_source_output_new_data_set_channel_map(&data, map);
- data.module = c->protocol->module;
- data.client = c->client;
if (!(source_output = pa_source_output_new(c->protocol->core, &data, 0)))
return NULL;
- s = pa_xnew(struct record_stream, 1);
+ s = pa_msgobject_new(record_stream);
+ c->parent.parent.free = record_stream_free;
+ c->parent.process_msg = record_stream_process_msg;
s->connection = c;
s->source_output = source_output;
s->source_output->push = source_output_push_cb;
s->memblockq = pa_memblockq_new(
0,
- maxlength,
+ *maxlength,
0,
base = pa_frame_size(ss),
1,
0,
NULL);
- assert(s->memblockq);
s->fragment_size = (fragment_size/base)*base;
- if (!s->fragment_size)
+ if (s->fragment_size <= 0)
s->fragment_size = base;
+ *maxlength = pa_memblockq_get_maxlength(s->memblockq);
pa_idxset_put(c->record_streams, s, &s->index);
+
+ pa_source_output_put(s->source_output);
return s;
}
-static void record_stream_free(struct record_stream* r) {
- assert(r && r->connection);
+static void playback_stream_unlink(playback_stream *s) {
+ pa_assert(s);
+
+ if (!s->connection)
+ return;
- pa_idxset_remove_by_data(r->connection->record_streams, r, NULL);
- pa_source_output_disconnect(r->source_output);
- pa_source_output_unref(r->source_output);
- pa_memblockq_free(r->memblockq);
- pa_xfree(r);
+ if (s->sink_input) {
+ pa_sink_input_disconnect(s->sink_input);
+ pa_sink_input_unref(s->sink_input);
+ s->sink_input = NULL;
+ }
+
+ if (s->drain_request)
+ pa_pstream_send_error(s->connection->pstream, s->drain_tag, PA_ERR_NOENTITY);
+
+ pa_assert_se(pa_idxset_remove_by_data(s->connection->output_streams, s, NULL) == s);
+ playback_stream_unref(s);
+ s->connection = NULL;
+}
+
+static void playback_stream_free(pa_object* o) {
+ playback_stream *s = PLAYBACK_STREAM(o);
+ pa_assert(s);
+
+ playback_stream_unlink(s);
+
+ pa_memblockq_free(s->memblockq);
+ pa_xfree(s);
}
-static struct playback_stream* playback_stream_new(
- struct connection *c,
+static int playback_stream_process_msg(pa_msgobject *o, int code, void*userdata, int64_t offset, pa_memchunk *chunk) {
+ playback_stream *s = PLAYBACK_STREAM(o);
+ playback_stream_assert_ref(s);
+
+ switch (code) {
+ case PLAYBACK_STREAM_MESSAGE_REQUEST_DATA: {
+ pa_tagstruct *t;
+ int32_t l;
+
+ if ((l = pa_atomic_load(&s->missing)) <= 0)
+ break;
+
+ pa_assert_se(pa_atomic_sub(&s->missing, l) >= l);
+
+ t = pa_tagstruct_new(NULL, 0);
+ pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
+ pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
+ pa_tagstruct_putu32(t, s->index);
+ pa_tagstruct_putu32(t, l);
+ pa_pstream_send_tagstruct(s->connection->pstream, t);
+
+ /* pa_log("Requesting %u bytes", l); */
+ break;
+ }
+
+ case PLAYBACK_STREAM_MESSAGE_UNDERFLOW: {
+ pa_tagstruct *t;
+
+ /* Report that we're empty */
+ t = pa_tagstruct_new(NULL, 0);
+ pa_tagstruct_putu32(t, PA_COMMAND_UNDERFLOW);
+ pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
+ pa_tagstruct_putu32(t, s->index);
+ pa_pstream_send_tagstruct(s->connection->pstream, t);
+ break;
+ }
+
+ case PLAYBACK_STREAM_MESSAGE_OVERFLOW: {
+ pa_tagstruct *t;
+
+ /* Notify the user we're overflowed*/
+ t = pa_tagstruct_new(NULL, 0);
+ pa_tagstruct_putu32(t, PA_COMMAND_OVERFLOW);
+ pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
+ pa_tagstruct_putu32(t, s->index);
+ pa_pstream_send_tagstruct(s->connection->pstream, t);
+ break;
+ }
+
+ case PLAYBACK_STREAM_MESSAGE_DRAIN_ACK:
+ pa_pstream_send_simple_ack(s->connection->pstream, PA_PTR_TO_UINT(userdata));
+ break;
+
+ }
+
+ return 0;
+}
+
+static playback_stream* playback_stream_new(
+ connection *c,
pa_sink *sink,
const pa_sample_spec *ss,
const pa_channel_map *map,
const char *name,
- size_t maxlength,
- size_t tlength,
- size_t prebuf,
- size_t minreq,
+ size_t *maxlength,
+ size_t *tlength,
+ size_t *prebuf,
+ size_t *minreq,
pa_cvolume *volume,
- uint32_t syncid) {
+ uint32_t syncid,
+ int corked,
+ size_t *missing) {
- struct playback_stream *s, *ssync;
+ playback_stream *s, *ssync;
pa_sink_input *sink_input;
pa_memblock *silence;
uint32_t idx;
int64_t start_index;
pa_sink_input_new_data data;
- assert(c && ss && name && maxlength);
+ pa_assert(c);
+ pa_assert(ss);
+ pa_assert(name);
+ pa_assert(maxlength);
/* Find syncid group */
for (ssync = pa_idxset_first(c->output_streams, &idx); ssync; ssync = pa_idxset_next(c->output_streams, &idx)) {
- if (ssync->type != PLAYBACK_STREAM)
+ if (!playback_stream_isinstance(ssync))
continue;
if (ssync->syncid == syncid)
}
/* Synced streams must connect to the same sink */
- if (ssync)
- sink = ssync->sink_input->sink;
+ if (ssync) {
+
+ if (!sink)
+ sink = ssync->sink_input->sink;
+ else if (sink != ssync->sink_input->sink)
+ return NULL;
+ }
pa_sink_input_new_data_init(&data);
data.sink = sink;
pa_sink_input_new_data_set_volume(&data, volume);
data.module = c->protocol->module;
data.client = c->client;
+ data.start_corked = corked;
+ data.sync_base = ssync ? ssync->sink_input : NULL;
if (!(sink_input = pa_sink_input_new(c->protocol->core, &data, 0)))
return NULL;
- s = pa_xnew(struct playback_stream, 1);
- s->type = PLAYBACK_STREAM;
+ s = pa_msgobject_new(playback_stream);
+ c->parent.parent.free = playback_stream_free;
+ c->parent.process_msg = playback_stream_process_msg;
s->connection = c;
s->syncid = syncid;
s->sink_input = sink_input;
s->underrun = 1;
+ s->sink_input->parent.process_msg = sink_input_process_msg;
s->sink_input->peek = sink_input_peek_cb;
s->sink_input->drop = sink_input_drop_cb;
s->sink_input->kill = sink_input_kill_cb;
- s->sink_input->get_latency = sink_input_get_latency_cb;
s->sink_input->userdata = s;
- if (ssync) {
- /* Sync id found, now find head of list */
- PA_LLIST_FIND_HEAD(struct playback_stream, ssync, &ssync);
-
- /* Prepend ourselves */
- PA_LLIST_PREPEND(struct playback_stream, ssync, s);
-
- /* Set our start index to the current read index of the other grozp member(s) */
- assert(ssync->next);
- start_index = pa_memblockq_get_read_index(ssync->next->memblockq);
- } else {
- /* This ia a new sync group */
- PA_LLIST_INIT(struct playback_stream, s);
- start_index = 0;
- }
+ start_index = ssync ? pa_memblockq_get_read_index(ssync->memblockq) : 0;
silence = pa_silence_memblock_new(c->protocol->core->mempool, ss, 0);
s->memblockq = pa_memblockq_new(
start_index,
- maxlength,
- tlength,
+ *maxlength,
+ *tlength,
pa_frame_size(ss),
- prebuf,
- minreq,
+ *prebuf,
+ *minreq,
silence);
pa_memblock_unref(silence);
- s->requested_bytes = 0;
+ *maxlength = pa_memblockq_get_maxlength(s->memblockq);
+ *tlength = pa_memblockq_get_tlength(s->memblockq);
+ *prebuf = pa_memblockq_get_prebuf(s->memblockq);
+ *minreq = pa_memblockq_get_minreq(s->memblockq);
+ *missing = pa_memblockq_missing(s->memblockq);
+
+ pa_atomic_store(&s->missing, 0);
+ s->last_missing = *missing;
s->drain_request = 0;
pa_idxset_put(c->output_streams, s, &s->index);
+ pa_sink_input_put(s->sink_input);
+
return s;
}
-static void playback_stream_free(struct playback_stream* p) {
- struct playback_stream *head;
- assert(p && p->connection);
+static void connection_unlink(connection *c) {
+ record_stream *r;
+ output_stream *o;
- if (p->drain_request)
- pa_pstream_send_error(p->connection->pstream, p->drain_tag, PA_ERR_NOENTITY);
+ pa_assert(c);
- PA_LLIST_FIND_HEAD(struct playback_stream, p, &head);
- PA_LLIST_REMOVE(struct playback_stream, head, p);
-
- pa_idxset_remove_by_data(p->connection->output_streams, p, NULL);
- pa_sink_input_disconnect(p->sink_input);
- pa_sink_input_unref(p->sink_input);
- pa_memblockq_free(p->memblockq);
- pa_xfree(p);
-}
-
-static void connection_free(struct connection *c) {
- struct record_stream *r;
- struct output_stream *o;
- assert(c && c->protocol);
+ if (!c->protocol)
+ return;
- pa_idxset_remove_by_data(c->protocol->connections, c, NULL);
while ((r = pa_idxset_first(c->record_streams, NULL)))
- record_stream_free(r);
- pa_idxset_free(c->record_streams, NULL, NULL);
+ record_stream_unlink(r);
while ((o = pa_idxset_first(c->output_streams, NULL)))
- if (o->type == PLAYBACK_STREAM)
- playback_stream_free((struct playback_stream*) o);
+ if (playback_stream_isinstance(o))
+ playback_stream_unlink(PLAYBACK_STREAM(o));
else
- upload_stream_free((struct upload_stream*) o);
- pa_idxset_free(c->output_streams, NULL, NULL);
-
- pa_pdispatch_unref(c->pdispatch);
- pa_pstream_close(c->pstream);
- pa_pstream_unref(c->pstream);
- pa_client_free(c->client);
+ upload_stream_unlink(UPLOAD_STREAM(o));
if (c->subscription)
pa_subscription_free(c->subscription);
- if (c->auth_timeout_event)
+ if (c->pstream)
+ pa_pstream_close(c->pstream);
+
+ if (c->auth_timeout_event) {
c->protocol->core->mainloop->time_free(c->auth_timeout_event);
+ c->auth_timeout_event = NULL;
+ }
+
+ pa_assert_se(pa_idxset_remove_by_data(c->protocol->connections, c, NULL) == c);
+ connection_unref(c);
+ c->protocol = NULL;
+}
+
+static void connection_free(pa_object *o) {
+ connection *c = CONNECTION(o);
+
+ pa_assert(c);
+
+ connection_unlink(c);
+
+ pa_idxset_free(c->record_streams, NULL, NULL);
+ pa_idxset_free(c->output_streams, NULL, NULL);
+
+ pa_pdispatch_unref(c->pdispatch);
+ pa_pstream_unref(c->pstream);
+ pa_client_free(c->client);
pa_xfree(c);
}
-static void request_bytes(struct playback_stream *s) {
- pa_tagstruct *t;
- size_t l;
- assert(s);
+static void request_bytes(playback_stream *s) {
+ size_t new_missing, delta, previous_missing;
- if (!(l = pa_memblockq_missing(s->memblockq)))
- return;
-
- if (l <= s->requested_bytes)
- return;
+ playback_stream_assert_ref(s);
- l -= s->requested_bytes;
+ new_missing = pa_memblockq_missing(s->memblockq);
- if (l < pa_memblockq_get_minreq(s->memblockq))
+ if (new_missing <= s->last_missing)
return;
- s->requested_bytes += l;
-
- t = pa_tagstruct_new(NULL, 0);
- assert(t);
- pa_tagstruct_putu32(t, PA_COMMAND_REQUEST);
- pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
- pa_tagstruct_putu32(t, s->index);
- pa_tagstruct_putu32(t, l);
- pa_pstream_send_tagstruct(s->connection->pstream, t);
+ delta = new_missing - s->last_missing;
+ s->last_missing = new_missing;
-/* pa_log("Requesting %u bytes", l); */
+ previous_missing = pa_atomic_add(&s->missing, delta);
+ if (previous_missing < pa_memblockq_get_minreq(s->memblockq) && previous_missing+delta >= pa_memblockq_get_minreq(s->memblockq))
+ pa_asyncmsgq_post(s->connection->protocol->core->asyncmsgq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_REQUEST_DATA, NULL, 0, NULL, NULL);
}
-static void send_memblock(struct connection *c) {
+static void send_memblock(connection *c) {
uint32_t start;
- struct record_stream *r;
+ record_stream *r;
start = PA_IDXSET_INVALID;
for (;;) {
pa_memchunk chunk;
- if (!(r = pa_idxset_rrobin(c->record_streams, &c->rrobin_index)))
+ if (!(r = RECORD_STREAM(pa_idxset_rrobin(c->record_streams, &c->rrobin_index))))
return;
if (start == PA_IDXSET_INVALID)
schunk.length = r->fragment_size;
pa_pstream_send_memblock(c->pstream, r->index, 0, PA_SEEK_RELATIVE, &schunk);
- pa_memblockq_drop(r->memblockq, &chunk, schunk.length);
+
+ pa_memblockq_drop(r->memblockq, schunk.length);
pa_memblock_unref(schunk.memblock);
return;
}
}
-static void send_playback_stream_killed(struct playback_stream *p) {
+static void send_playback_stream_killed(playback_stream *p) {
pa_tagstruct *t;
- assert(p);
+ playback_stream_assert_ref(p);
t = pa_tagstruct_new(NULL, 0);
pa_tagstruct_putu32(t, PA_COMMAND_PLAYBACK_STREAM_KILLED);
pa_pstream_send_tagstruct(p->connection->pstream, t);
}
-static void send_record_stream_killed(struct record_stream *r) {
+static void send_record_stream_killed(record_stream *r) {
pa_tagstruct *t;
- assert(r);
+ record_stream_assert_ref(r);
t = pa_tagstruct_new(NULL, 0);
pa_tagstruct_putu32(t, PA_COMMAND_RECORD_STREAM_KILLED);
/*** sinkinput callbacks ***/
-static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) {
- struct playback_stream *s;
- assert(i && i->userdata && chunk);
- s = i->userdata;
+static int sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk) {
+ pa_sink_input *i = PA_SINK_INPUT(o);
+ playback_stream *s;
- if (pa_memblockq_get_length(s->memblockq) <= 0 && !s->underrun) {
- pa_tagstruct *t;
+ pa_sink_input_assert_ref(i);
+ s = PLAYBACK_STREAM(i->userdata);
+ playback_stream_assert_ref(s);
+
+ switch (code) {
+
+ case SINK_INPUT_MESSAGE_SEEK:
+ pa_memblockq_seek(s->memblockq, offset, PA_PTR_TO_UINT(userdata));
+ return 0;
+
+ case SINK_INPUT_MESSAGE_POST_DATA: {
+ pa_assert(chunk);
+
+ if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
+
+ pa_log_warn("Failed to push data into queue");
+ pa_asyncmsgq_post(s->connection->protocol->core->asyncmsgq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_OVERFLOW, NULL, 0, NULL, NULL);
+ pa_memblockq_seek(s->memblockq, chunk->length, PA_SEEK_RELATIVE);
+ }
+
+ s->underrun = 0;
+ return 0;
+ }
+
+ case SINK_INPUT_MESSAGE_DRAIN: {
+
+ pa_memblockq_prebuf_disable(s->memblockq);
+
+ if (!pa_memblockq_is_readable(s->memblockq))
+ pa_asyncmsgq_post(s->connection->protocol->core->asyncmsgq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_DRAIN_ACK, userdata, 0, NULL, NULL);
+ else {
+ s->drain_tag = PA_PTR_TO_UINT(userdata);
+ s->drain_request = 1;
+ }
+
+ return 0;
+ }
+
+ case SINK_INPUT_MESSAGE_FLUSH:
+ case SINK_INPUT_MESSAGE_PREBUF_FORCE:
+ case SINK_INPUT_MESSAGE_TRIGGER: {
+
+ pa_sink_input *isync;
+ void (*func)(pa_memblockq *bq);
+
+ switch (code) {
+ case SINK_INPUT_MESSAGE_FLUSH:
+ func = pa_memblockq_flush;
+ break;
+
+ case SINK_INPUT_MESSAGE_PREBUF_FORCE:
+ func = pa_memblockq_prebuf_force;
+ break;
+
+ case SINK_INPUT_MESSAGE_TRIGGER:
+ func = pa_memblockq_prebuf_disable;
+ break;
+
+ default:
+ pa_assert_not_reached();
+ }
+
+ func(s->memblockq);
+ s->underrun = 0;
+ request_bytes(s);
+
+ /* Do the same for all other members in the sync group */
+ for (isync = i->sync_prev; isync; isync = isync->sync_prev) {
+ playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
+ func(ssync->memblockq);
+ ssync->underrun = 0;
+ request_bytes(ssync);
+ }
+
+ for (isync = i->sync_next; isync; isync = isync->sync_next) {
+ playback_stream *ssync = PLAYBACK_STREAM(isync->userdata);
+ func(ssync->memblockq);
+ ssync->underrun = 0;
+ request_bytes(ssync);
+ }
+
+ return 0;
+ }
+
+ case PA_SINK_INPUT_MESSAGE_SET_STATE:
+
+ pa_memblockq_prebuf_force(s->memblockq);
+ break;
+
+ case PA_SINK_INPUT_MESSAGE_GET_LATENCY: {
+ pa_usec_t *r = userdata;
+
+ *r = pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &i->sample_spec);
+
+ /* Fall through, the default handler will add in the extra
+ * latency added by the resampler */
+ break;
+ }
+ }
- /* Report that we're empty */
+ return pa_sink_input_process_msg(o, code, userdata, offset, chunk);
+}
+
+static int sink_input_peek_cb(pa_sink_input *i, pa_memchunk *chunk) {
+ playback_stream *s;
- t = pa_tagstruct_new(NULL, 0);
- pa_tagstruct_putu32(t, PA_COMMAND_UNDERFLOW);
- pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
- pa_tagstruct_putu32(t, s->index);
- pa_pstream_send_tagstruct(s->connection->pstream, t);
+ pa_sink_input_assert_ref(i);
+ s = PLAYBACK_STREAM(i->userdata);
+ playback_stream_assert_ref(s);
+ pa_assert(chunk);
+ if (pa_memblockq_get_length(s->memblockq) <= 0 && !s->underrun) {
+ pa_asyncmsgq_post(s->connection->protocol->core->asyncmsgq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_UNDERFLOW, NULL, 0, NULL, NULL);
s->underrun = 1;
}
return 0;
}
-static void sink_input_drop_cb(pa_sink_input *i, const pa_memchunk *chunk, size_t length) {
- struct playback_stream *s;
- assert(i && i->userdata && length);
- s = i->userdata;
+static void sink_input_drop_cb(pa_sink_input *i, size_t length) {
+ playback_stream *s;
- pa_memblockq_drop(s->memblockq, chunk, length);
+ pa_sink_input_assert_ref(i);
+ s = PLAYBACK_STREAM(i->userdata);
+ playback_stream_assert_ref(s);
+ pa_assert(length > 0);
- request_bytes(s);
+ pa_memblockq_drop(s->memblockq, length);
if (s->drain_request && !pa_memblockq_is_readable(s->memblockq)) {
- pa_pstream_send_simple_ack(s->connection->pstream, s->drain_tag);
s->drain_request = 0;
+ pa_asyncmsgq_post(s->connection->protocol->core->asyncmsgq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_DRAIN_ACK, PA_UINT_TO_PTR(s->drain_tag), 0, NULL, NULL);
}
+ request_bytes(s);
+
/* pa_log("after_drop: %u %u", pa_memblockq_get_length(s->memblockq), pa_memblockq_is_readable(s->memblockq)); */
}
static void sink_input_kill_cb(pa_sink_input *i) {
- assert(i && i->userdata);
- send_playback_stream_killed((struct playback_stream *) i->userdata);
- playback_stream_free((struct playback_stream *) i->userdata);
-}
+ playback_stream *s;
-static pa_usec_t sink_input_get_latency_cb(pa_sink_input *i) {
- struct playback_stream *s;
- assert(i && i->userdata);
- s = i->userdata;
+ pa_sink_input_assert_ref(i);
+ s = PLAYBACK_STREAM(i->userdata);
+ playback_stream_assert_ref(s);
- /*pa_log("get_latency: %u", pa_memblockq_get_length(s->memblockq));*/
-
- return pa_bytes_to_usec(pa_memblockq_get_length(s->memblockq), &s->sink_input->sample_spec);
+ send_playback_stream_killed(s);
+ playback_stream_unlink(s);
}
/*** source_output callbacks ***/
static void source_output_push_cb(pa_source_output *o, const pa_memchunk *chunk) {
- struct record_stream *s;
- assert(o && o->userdata && chunk);
- s = o->userdata;
+ record_stream *s;
- if (pa_memblockq_push_align(s->memblockq, chunk) < 0) {
- pa_log_warn("Failed to push data into output queue.");
- return;
- }
+ pa_source_output_assert_ref(o);
+ s = RECORD_STREAM(o->userdata);
+ record_stream_assert_ref(s);
+ pa_assert(chunk);
- if (!pa_pstream_is_pending(s->connection->pstream))
- send_memblock(s->connection);
+ pa_asyncmsgq_post(s->connection->protocol->core->asyncmsgq, PA_MSGOBJECT(s), RECORD_STREAM_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
}
static void source_output_kill_cb(pa_source_output *o) {
- assert(o && o->userdata);
- send_record_stream_killed((struct record_stream *) o->userdata);
- record_stream_free((struct record_stream *) o->userdata);
+ record_stream *s;
+
+ pa_source_output_assert_ref(o);
+ s = RECORD_STREAM(o->userdata);
+ record_stream_assert_ref(s);
+
+ send_record_stream_killed(s);
+ record_stream_unlink(s);
}
static pa_usec_t source_output_get_latency_cb(pa_source_output *o) {
- struct record_stream *s;
- assert(o && o->userdata);
- s = o->userdata;
+ record_stream *s;
+
+ pa_source_output_assert_ref(o);
+ s = RECORD_STREAM(o->userdata);
+ record_stream_assert_ref(s);
/*pa_log("get_latency: %u", pa_memblockq_get_length(s->memblockq));*/
/*** pdispatch callbacks ***/
-static void protocol_error(struct connection *c) {
+static void protocol_error(connection *c) {
pa_log("protocol error, kicking client");
- connection_free(c);
+ connection_unlink(c);
}
#define CHECK_VALIDITY(pstream, expression, tag, error) do { \
}
static void command_create_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
- struct playback_stream *s;
- uint32_t maxlength, tlength, prebuf, minreq, sink_index, syncid;
+ connection *c = CONNECTION(userdata);
+ playback_stream *s;
+ uint32_t maxlength, tlength, prebuf, minreq, sink_index, syncid, missing;
const char *name, *sink_name;
pa_sample_spec ss;
pa_channel_map map;
pa_sink *sink = NULL;
pa_cvolume volume;
int corked;
-
- assert(c && t && c->protocol && c->protocol->core);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_get(
t,
CHECK_VALIDITY(c->pstream, sink, tag, PA_ERR_NOENTITY);
}
- s = playback_stream_new(c, sink, &ss, &map, name, maxlength, tlength, prebuf, minreq, &volume, syncid);
+ s = playback_stream_new(c, sink, &ss, &map, name, &maxlength, &tlength, &prebuf, &minreq, &volume, syncid, corked, &missing);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
- pa_sink_input_cork(s->sink_input, corked);
-
reply = reply_new(tag);
pa_tagstruct_putu32(reply, s->index);
- assert(s->sink_input);
+ pa_assert(s->sink_input);
pa_tagstruct_putu32(reply, s->sink_input->index);
- pa_tagstruct_putu32(reply, s->requested_bytes = pa_memblockq_missing(s->memblockq));
+ pa_tagstruct_putu32(reply, missing);
if (c->version >= 9) {
/* Since 0.9 we support sending the buffer metrics back to the client */
- pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_maxlength(s->memblockq));
- pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_tlength(s->memblockq));
- pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_prebuf(s->memblockq));
- pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_minreq(s->memblockq));
+ pa_tagstruct_putu32(reply, (uint32_t) maxlength);
+ pa_tagstruct_putu32(reply, (uint32_t) tlength);
+ pa_tagstruct_putu32(reply, (uint32_t) prebuf);
+ pa_tagstruct_putu32(reply, (uint32_t) minreq);
}
pa_pstream_send_tagstruct(c->pstream, reply);
- request_bytes(s);
}
static void command_delete_stream(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t channel;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &channel) < 0 ||
!pa_tagstruct_eof(t)) {
CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
- if (command == PA_COMMAND_DELETE_PLAYBACK_STREAM) {
- struct playback_stream *s;
- if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != PLAYBACK_STREAM)) {
- pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
- return;
+ switch (command) {
+
+ case PA_COMMAND_DELETE_PLAYBACK_STREAM: {
+ playback_stream *s;
+ if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !playback_stream_isinstance(s)) {
+ pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
+ return;
+ }
+
+ playback_stream_unlink(s);
+ break;
}
-
- playback_stream_free(s);
- } else if (command == PA_COMMAND_DELETE_RECORD_STREAM) {
- struct record_stream *s;
- if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
- pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
- return;
+
+ case PA_COMMAND_DELETE_RECORD_STREAM: {
+ record_stream *s;
+ if (!(s = pa_idxset_get_by_index(c->record_streams, channel))) {
+ pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
+ return;
+ }
+
+ record_stream_unlink(s);
+ break;
}
- record_stream_free(s);
- } else {
- struct upload_stream *s;
- assert(command == PA_COMMAND_DELETE_UPLOAD_STREAM);
- if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || (s->type != UPLOAD_STREAM)) {
- pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
- return;
+ case PA_COMMAND_DELETE_UPLOAD_STREAM: {
+ upload_stream *s;
+
+ if (!(s = pa_idxset_get_by_index(c->output_streams, channel)) || !upload_stream_isinstance(s)) {
+ pa_pstream_send_error(c->pstream, tag, PA_ERR_EXIST);
+ return;
+ }
+
+ upload_stream_unlink(s);
+ break;
}
- upload_stream_free(s);
+ default:
+ pa_assert_not_reached();
}
pa_pstream_send_simple_ack(c->pstream, tag);
}
static void command_create_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
- struct record_stream *s;
+ connection *c = CONNECTION(userdata);
+ record_stream *s;
uint32_t maxlength, fragment_size;
uint32_t source_index;
const char *name, *source_name;
pa_tagstruct *reply;
pa_source *source = NULL;
int corked;
- assert(c && t && c->protocol && c->protocol->core);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &name) < 0 ||
pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
CHECK_VALIDITY(c->pstream, source, tag, PA_ERR_NOENTITY);
}
- s = record_stream_new(c, source, &ss, &map, name, maxlength, fragment_size);
+ s = record_stream_new(c, source, &ss, &map, name, &maxlength, fragment_size, corked);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_INVALID);
- pa_source_output_cork(s->source_output, corked);
-
reply = reply_new(tag);
pa_tagstruct_putu32(reply, s->index);
- assert(s->source_output);
+ pa_assert(s->source_output);
pa_tagstruct_putu32(reply, s->source_output->index);
if (c->version >= 9) {
/* Since 0.9 we support sending the buffer metrics back to the client */
- pa_tagstruct_putu32(reply, (uint32_t) pa_memblockq_get_maxlength(s->memblockq));
+ pa_tagstruct_putu32(reply, (uint32_t) maxlength);
pa_tagstruct_putu32(reply, (uint32_t) s->fragment_size);
}
}
static void command_exit(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
- assert(c && t);
+ connection *c = CONNECTION(userdata);
+ connection_assert_ref(c);
+ pa_assert(t);
+
if (!pa_tagstruct_eof(t)) {
protocol_error(c);
return;
CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
- assert(c->protocol && c->protocol->core && c->protocol->core->mainloop);
c->protocol->core->mainloop->quit(c->protocol->core->mainloop, 0);
pa_pstream_send_simple_ack(c->pstream, tag); /* nonsense */
}
static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const void*cookie;
pa_tagstruct *reply;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &c->version) < 0 ||
pa_tagstruct_get_arbitrary(t, &cookie, PA_NATIVE_COOKIE_LENGTH) < 0 ||
}
static void command_set_client_name(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const char *name;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &name) < 0 ||
!pa_tagstruct_eof(t)) {
}
static void command_lookup(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const char *name;
uint32_t idx = PA_IDXSET_INVALID;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &name) < 0 ||
!pa_tagstruct_eof(t)) {
idx = sink->index;
} else {
pa_source *source;
- assert(command == PA_COMMAND_LOOKUP_SOURCE);
+ pa_assert(command == PA_COMMAND_LOOKUP_SOURCE);
if ((source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1)))
idx = source->index;
}
}
static void command_drain_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
- struct playback_stream *s;
- assert(c && t);
+ playback_stream *s;
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
!pa_tagstruct_eof(t)) {
CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
s = pa_idxset_get_by_index(c->output_streams, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == PLAYBACK_STREAM, tag, PA_ERR_NOENTITY);
-
- s->drain_request = 0;
-
- pa_memblockq_prebuf_disable(s->memblockq);
+ CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
- if (!pa_memblockq_is_readable(s->memblockq)) {
-/* pa_log("immediate drain: %u", pa_memblockq_get_length(s->memblockq)); */
- pa_pstream_send_simple_ack(c->pstream, tag);
- } else {
-/* pa_log("slow drain triggered"); */
- s->drain_request = 1;
- s->drain_tag = tag;
-
- pa_sink_notify(s->sink_input->sink);
- }
+ pa_asyncmsgq_post(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_DRAIN, PA_UINT_TO_PTR(tag), 0, NULL, NULL);
}
static void command_stat(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_tagstruct *reply;
const pa_mempool_stat *stat;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (!pa_tagstruct_eof(t)) {
protocol_error(c);
}
static void command_get_playback_latency(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_tagstruct *reply;
- struct playback_stream *s;
+ playback_stream *s;
struct timeval tv, now;
uint32_t idx;
pa_usec_t latency;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
pa_tagstruct_get_timeval(t, &tv) < 0 ||
CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
s = pa_idxset_get_by_index(c->output_streams, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == PLAYBACK_STREAM, tag, PA_ERR_NOENTITY);
+ CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
reply = reply_new(tag);
latency = pa_sink_get_latency(s->sink_input->sink);
- if (s->sink_input->resampled_chunk.memblock)
- latency += pa_bytes_to_usec(s->sink_input->resampled_chunk.length, &s->sink_input->sample_spec);
+/* if (s->sink_input->resampled_chunk.memblock) */ /* FIXME*/
+/* latency += pa_bytes_to_usec(s->sink_input->resampled_chunk.length, &s->sink_input->sample_spec); */
pa_tagstruct_put_usec(reply, latency);
pa_tagstruct_put_usec(reply, 0);
}
static void command_get_record_latency(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_tagstruct *reply;
- struct record_stream *s;
+ record_stream *s;
struct timeval tv, now;
uint32_t idx;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
pa_tagstruct_get_timeval(t, &tv) < 0 ||
}
static void command_create_upload_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
- struct upload_stream *s;
+ connection *c = CONNECTION(userdata);
+ upload_stream *s;
uint32_t length;
const char *name;
pa_sample_spec ss;
pa_channel_map map;
pa_tagstruct *reply;
- assert(c && t && c->protocol && c->protocol->core);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &name) < 0 ||
pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
}
static void command_finish_upload_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t channel;
- struct upload_stream *s;
+ upload_stream *s;
uint32_t idx;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &channel) < 0 ||
!pa_tagstruct_eof(t)) {
s = pa_idxset_get_by_index(c->output_streams, channel);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == UPLOAD_STREAM, tag, PA_ERR_NOENTITY);
+ CHECK_VALIDITY(c->pstream, upload_stream_isinstance(s), tag, PA_ERR_NOENTITY);
if (pa_scache_add_item(c->protocol->core, s->name, &s->sample_spec, &s->channel_map, &s->memchunk, &idx) < 0)
pa_pstream_send_error(c->pstream, tag, PA_ERR_INTERNAL);
else
pa_pstream_send_simple_ack(c->pstream, tag);
- upload_stream_free(s);
+ upload_stream_unlink(s);
}
static void command_play_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t sink_index;
pa_volume_t volume;
pa_sink *sink;
const char *name, *sink_name;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &sink_index) < 0 ||
pa_tagstruct_gets(t, &sink_name) < 0 ||
}
static void command_remove_sample(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const char *name;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &name) < 0 ||
!pa_tagstruct_eof(t)) {
}
static void sink_fill_tagstruct(pa_tagstruct *t, pa_sink *sink) {
- assert(t && sink);
+ pa_assert(t);
+ pa_sink_assert_ref(sink);
+
pa_tagstruct_put(
t,
PA_TAG_U32, sink->index,
PA_TAG_STRING, sink->description,
PA_TAG_SAMPLE_SPEC, &sink->sample_spec,
PA_TAG_CHANNEL_MAP, &sink->channel_map,
- PA_TAG_U32, sink->owner ? sink->owner->index : PA_INVALID_INDEX,
- PA_TAG_CVOLUME, pa_sink_get_volume(sink, PA_MIXER_HARDWARE),
- PA_TAG_BOOLEAN, pa_sink_get_mute(sink, PA_MIXER_HARDWARE),
+ PA_TAG_U32, sink->module ? sink->module->index : PA_INVALID_INDEX,
+ PA_TAG_CVOLUME, pa_sink_get_volume(sink),
+ PA_TAG_BOOLEAN, pa_sink_get_mute(sink),
PA_TAG_U32, sink->monitor_source ? sink->monitor_source->index : PA_INVALID_INDEX,
PA_TAG_STRING, sink->monitor_source ? sink->monitor_source->name : NULL,
PA_TAG_USEC, pa_sink_get_latency(sink),
PA_TAG_STRING, sink->driver,
PA_TAG_U32,
- (sink->get_hw_volume ? PA_SINK_HW_VOLUME_CTRL : 0) |
- (sink->get_latency ? PA_SINK_LATENCY : 0) |
+ (sink->get_volume ? PA_SINK_HW_VOLUME_CTRL : 0) | /* FIXME */
+ (sink->get_latency ? PA_SINK_LATENCY : 0) | /* FIXME */
(sink->is_hardware ? PA_SINK_HARDWARE : 0),
PA_TAG_INVALID);
}
static void source_fill_tagstruct(pa_tagstruct *t, pa_source *source) {
- assert(t && source);
+ pa_assert(t);
+ pa_source_assert_ref(source);
+
pa_tagstruct_put(
t,
PA_TAG_U32, source->index,
PA_TAG_STRING, source->description,
PA_TAG_SAMPLE_SPEC, &source->sample_spec,
PA_TAG_CHANNEL_MAP, &source->channel_map,
- PA_TAG_U32, source->owner ? source->owner->index : PA_INVALID_INDEX,
- PA_TAG_CVOLUME, pa_source_get_volume(source, PA_MIXER_HARDWARE),
- PA_TAG_BOOLEAN, pa_source_get_mute(source, PA_MIXER_HARDWARE),
+ PA_TAG_U32, source->module ? source->module->index : PA_INVALID_INDEX,
+ PA_TAG_CVOLUME, pa_source_get_volume(source),
+ PA_TAG_BOOLEAN, pa_source_get_mute(source),
PA_TAG_U32, source->monitor_of ? source->monitor_of->index : PA_INVALID_INDEX,
PA_TAG_STRING, source->monitor_of ? source->monitor_of->name : NULL,
PA_TAG_USEC, pa_source_get_latency(source),
PA_TAG_STRING, source->driver,
PA_TAG_U32,
- (source->get_hw_volume ? PA_SOURCE_HW_VOLUME_CTRL : 0) |
- (source->get_latency ? PA_SOURCE_LATENCY : 0) |
+ (source->get_volume ? PA_SOURCE_HW_VOLUME_CTRL : 0) | /* FIXME */
+ (source->get_latency ? PA_SOURCE_LATENCY : 0) | /* FIXME */
(source->is_hardware ? PA_SOURCE_HARDWARE : 0),
PA_TAG_INVALID);
}
static void client_fill_tagstruct(pa_tagstruct *t, pa_client *client) {
- assert(t && client);
+ pa_assert(t);
+ pa_assert(client);
+
pa_tagstruct_putu32(t, client->index);
pa_tagstruct_puts(t, client->name);
pa_tagstruct_putu32(t, client->owner ? client->owner->index : PA_INVALID_INDEX);
}
static void module_fill_tagstruct(pa_tagstruct *t, pa_module *module) {
- assert(t && module);
+ pa_assert(t);
+ pa_assert(module);
+
pa_tagstruct_putu32(t, module->index);
pa_tagstruct_puts(t, module->name);
pa_tagstruct_puts(t, module->argument);
}
static void sink_input_fill_tagstruct(pa_tagstruct *t, pa_sink_input *s) {
- assert(t && s);
+ pa_assert(t);
+ pa_sink_input_assert_ref(s);
+
pa_tagstruct_putu32(t, s->index);
pa_tagstruct_puts(t, s->name);
pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
}
static void source_output_fill_tagstruct(pa_tagstruct *t, pa_source_output *s) {
- assert(t && s);
+ pa_assert(t);
+ pa_source_output_assert_ref(s);
+
pa_tagstruct_putu32(t, s->index);
pa_tagstruct_puts(t, s->name);
pa_tagstruct_putu32(t, s->module ? s->module->index : PA_INVALID_INDEX);
}
static void scache_fill_tagstruct(pa_tagstruct *t, pa_scache_entry *e) {
- assert(t && e);
+ pa_assert(t);
+ pa_assert(e);
+
pa_tagstruct_putu32(t, e->index);
pa_tagstruct_puts(t, e->name);
pa_tagstruct_put_cvolume(t, &e->volume);
}
static void command_get_info(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
pa_sink *sink = NULL;
pa_source *source = NULL;
pa_scache_entry *sce = NULL;
const char *name;
pa_tagstruct *reply;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
(command != PA_COMMAND_GET_CLIENT_INFO &&
else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO)
so = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
else {
- assert(command == PA_COMMAND_GET_SAMPLE_INFO);
+ pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO);
if (idx != PA_INVALID_INDEX)
sce = pa_idxset_get_by_index(c->protocol->core->scache, idx);
else
}
static void command_get_info_list(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_idxset *i;
uint32_t idx;
void *p;
pa_tagstruct *reply;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (!pa_tagstruct_eof(t)) {
protocol_error(c);
else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
i = c->protocol->core->source_outputs;
else {
- assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
+ pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
i = c->protocol->core->scache;
}
else if (command == PA_COMMAND_GET_SOURCE_OUTPUT_INFO_LIST)
source_output_fill_tagstruct(reply, p);
else {
- assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
+ pa_assert(command == PA_COMMAND_GET_SAMPLE_INFO_LIST);
scache_fill_tagstruct(reply, p);
}
}
}
static void command_get_server_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_tagstruct *reply;
char txt[256];
const char *n;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (!pa_tagstruct_eof(t)) {
protocol_error(c);
static void subscription_cb(pa_core *core, pa_subscription_event_type_t e, uint32_t idx, void *userdata) {
pa_tagstruct *t;
- struct connection *c = userdata;
- assert(c && core);
+ connection *c = CONNECTION(userdata);
+
+ connection_assert_ref(c);
+ pa_assert(t);
t = pa_tagstruct_new(NULL, 0);
pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE_EVENT);
}
static void command_subscribe(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_subscription_mask_t m;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &m) < 0 ||
!pa_tagstruct_eof(t)) {
if (m != 0) {
c->subscription = pa_subscription_new(c->protocol->core, m, subscription_cb, c);
- assert(c->subscription);
+ pa_assert(c->subscription);
} else
c->subscription = NULL;
pa_tagstruct *t,
void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
pa_cvolume volume;
pa_sink *sink = NULL;
pa_source *source = NULL;
pa_sink_input *si = NULL;
const char *name = NULL;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
(command == PA_COMMAND_SET_SINK_VOLUME && pa_tagstruct_gets(t, &name) < 0) ||
CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
CHECK_VALIDITY(c->pstream, pa_cvolume_valid(&volume), tag, PA_ERR_INVALID);
- if (command == PA_COMMAND_SET_SINK_VOLUME) {
- if (idx != PA_INVALID_INDEX)
- sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
- else
- sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
- } else if (command == PA_COMMAND_SET_SOURCE_VOLUME) {
- if (idx != (uint32_t) -1)
- source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
- else
- source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
- } else {
- assert(command == PA_COMMAND_SET_SINK_INPUT_VOLUME);
- si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
+ switch (command) {
+
+ case PA_COMMAND_SET_SINK_VOLUME:
+ if (idx != PA_INVALID_INDEX)
+ sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
+ else
+ sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
+ break;
+
+ case PA_COMMAND_SET_SOURCE_VOLUME:
+ if (idx != PA_INVALID_INDEX)
+ source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
+ else
+ source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
+ break;
+
+ case PA_COMMAND_SET_SINK_INPUT_VOLUME:
+ si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
+ break;
+
+ default:
+ pa_assert_not_reached();
}
CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
if (sink)
- pa_sink_set_volume(sink, PA_MIXER_HARDWARE, &volume);
+ pa_sink_set_volume(sink, &volume);
else if (source)
- pa_source_set_volume(source, PA_MIXER_HARDWARE, &volume);
+ pa_source_set_volume(source, &volume);
else if (si)
pa_sink_input_set_volume(si, &volume);
pa_tagstruct *t,
void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
int mute;
pa_sink *sink = NULL;
pa_source *source = NULL;
+ pa_sink_input *si = NULL;
const char *name = NULL;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
- pa_tagstruct_gets(t, &name) < 0 ||
+ (command == PA_COMMAND_SET_SINK_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
+ (command == PA_COMMAND_SET_SOURCE_MUTE && pa_tagstruct_gets(t, &name) < 0) ||
pa_tagstruct_get_boolean(t, &mute) ||
!pa_tagstruct_eof(t)) {
protocol_error(c);
CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX || !name || (*name && pa_utf8_valid(name)), tag, PA_ERR_INVALID);
- if (command == PA_COMMAND_SET_SINK_MUTE) {
- if (idx != PA_INVALID_INDEX)
- sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
- else
- sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
- } else {
- assert(command == PA_COMMAND_SET_SOURCE_MUTE);
- if (idx != (uint32_t) -1)
- source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
- else
- source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
+ switch (command) {
+
+ case PA_COMMAND_SET_SINK_MUTE:
+
+ if (idx != PA_INVALID_INDEX)
+ sink = pa_idxset_get_by_index(c->protocol->core->sinks, idx);
+ else
+ sink = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SINK, 1);
+
+ break;
+
+ case PA_COMMAND_SET_SOURCE_MUTE:
+ if (idx != PA_INVALID_INDEX)
+ source = pa_idxset_get_by_index(c->protocol->core->sources, idx);
+ else
+ source = pa_namereg_get(c->protocol->core, name, PA_NAMEREG_SOURCE, 1);
+
+ break;
+
+ case PA_COMMAND_SET_SINK_INPUT_MUTE:
+ si = pa_idxset_get_by_index(c->protocol->core->sink_inputs, idx);
+ break;
+
+ default:
+ pa_assert_not_reached();
}
- CHECK_VALIDITY(c->pstream, sink || source, tag, PA_ERR_NOENTITY);
+ CHECK_VALIDITY(c->pstream, si || sink || source, tag, PA_ERR_NOENTITY);
if (sink)
- pa_sink_set_mute(sink, PA_MIXER_HARDWARE, mute);
+ pa_sink_set_mute(sink, mute);
else if (source)
- pa_source_set_mute(source, PA_MIXER_HARDWARE, mute);
+ pa_source_set_mute(source, mute);
+ else if (si)
+ pa_sink_input_set_mute(si, mute);
pa_pstream_send_simple_ack(c->pstream, tag);
}
static void command_cork_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
int b;
- struct playback_stream *s, *ssync;
- assert(c && t);
+ playback_stream *s;
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
pa_tagstruct_get_boolean(t, &b) < 0 ||
CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
s = pa_idxset_get_by_index(c->output_streams, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == PLAYBACK_STREAM, tag, PA_ERR_NOENTITY);
+ CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
pa_sink_input_cork(s->sink_input, b);
- pa_memblockq_prebuf_force(s->memblockq);
-
- /* Do the same for all other members in the sync group */
- for (ssync = s->prev; ssync; ssync = ssync->prev) {
- pa_sink_input_cork(ssync->sink_input, b);
- pa_memblockq_prebuf_force(ssync->memblockq);
- }
-
- for (ssync = s->next; ssync; ssync = ssync->next) {
- pa_sink_input_cork(ssync->sink_input, b);
- pa_memblockq_prebuf_force(ssync->memblockq);
- }
-
pa_pstream_send_simple_ack(c->pstream, tag);
}
-static void command_flush_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+static void command_trigger_or_flush_or_prebuf_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
+ connection *c = CONNECTION(userdata);
uint32_t idx;
- struct playback_stream *s, *ssync;
- assert(c && t);
-
- if (pa_tagstruct_getu32(t, &idx) < 0 ||
- !pa_tagstruct_eof(t)) {
- protocol_error(c);
- return;
- }
-
- CHECK_VALIDITY(c->pstream, c->authorized, tag, PA_ERR_ACCESS);
- CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
- s = pa_idxset_get_by_index(c->output_streams, idx);
- CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == PLAYBACK_STREAM, tag, PA_ERR_NOENTITY);
-
- pa_memblockq_flush(s->memblockq);
- s->underrun = 0;
+ playback_stream *s;
- /* Do the same for all other members in the sync group */
- for (ssync = s->prev; ssync; ssync = ssync->prev) {
- pa_memblockq_flush(ssync->memblockq);
- ssync->underrun = 0;
- }
-
- for (ssync = s->next; ssync; ssync = ssync->next) {
- pa_memblockq_flush(ssync->memblockq);
- ssync->underrun = 0;
- }
-
- pa_pstream_send_simple_ack(c->pstream, tag);
- pa_sink_notify(s->sink_input->sink);
- request_bytes(s);
-
- for (ssync = s->prev; ssync; ssync = ssync->prev)
- request_bytes(ssync);
-
- for (ssync = s->next; ssync; ssync = ssync->next)
- request_bytes(ssync);
-}
-
-static void command_trigger_or_prebuf_playback_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
- uint32_t idx;
- struct playback_stream *s;
- assert(c && t);
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
!pa_tagstruct_eof(t)) {
CHECK_VALIDITY(c->pstream, idx != PA_INVALID_INDEX, tag, PA_ERR_INVALID);
s = pa_idxset_get_by_index(c->output_streams, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == PLAYBACK_STREAM, tag, PA_ERR_NOENTITY);
+ CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
switch (command) {
+ case PA_COMMAND_FLUSH_PLAYBACK_STREAM:
+ pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_FLUSH, NULL, 0, NULL);
+ break;
+
case PA_COMMAND_PREBUF_PLAYBACK_STREAM:
- pa_memblockq_prebuf_force(s->memblockq);
+ pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_TRIGGER, NULL, 0, NULL);
break;
case PA_COMMAND_TRIGGER_PLAYBACK_STREAM:
- pa_memblockq_prebuf_disable(s->memblockq);
+ pa_asyncmsgq_send(s->sink_input->sink->asyncmsgq, PA_MSGOBJECT(s->sink_input), SINK_INPUT_MESSAGE_PREBUF_FORCE, NULL, 0, NULL);
break;
default:
- abort();
+ pa_assert_not_reached();
}
- pa_sink_notify(s->sink_input->sink);
pa_pstream_send_simple_ack(c->pstream, tag);
- request_bytes(s);
}
static void command_cork_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
- struct record_stream *s;
+ record_stream *s;
int b;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
pa_tagstruct_get_boolean(t, &b) < 0 ||
}
static void command_flush_record_stream(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
- struct record_stream *s;
- assert(c && t);
+ record_stream *s;
+ connection_assert_ref(c);
+ pa_assert(t);
+
if (pa_tagstruct_getu32(t, &idx) < 0 ||
!pa_tagstruct_eof(t)) {
protocol_error(c);
}
static void command_set_default_sink_or_source(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const char *s;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &s) < 0 ||
!pa_tagstruct_eof(t)) {
}
static void command_set_stream_name(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
const char *name;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
pa_tagstruct_gets(t, &name) < 0 ||
CHECK_VALIDITY(c->pstream, name && pa_utf8_valid(name), tag, PA_ERR_INVALID);
if (command == PA_COMMAND_SET_PLAYBACK_STREAM_NAME) {
- struct playback_stream *s;
+ playback_stream *s;
s = pa_idxset_get_by_index(c->output_streams, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
- CHECK_VALIDITY(c->pstream, s->type == PLAYBACK_STREAM, tag, PA_ERR_NOENTITY);
+ CHECK_VALIDITY(c->pstream, playback_stream_isinstance(s), tag, PA_ERR_NOENTITY);
pa_sink_input_set_name(s->sink_input, name);
} else {
- struct record_stream *s;
+ record_stream *s;
s = pa_idxset_get_by_index(c->record_streams, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
}
static void command_kill(PA_GCC_UNUSED pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
!pa_tagstruct_eof(t)) {
} else {
pa_source_output *s;
- assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
+ pa_assert(command == PA_COMMAND_KILL_SOURCE_OUTPUT);
s = pa_idxset_get_by_index(c->protocol->core->source_outputs, idx);
CHECK_VALIDITY(c->pstream, s, tag, PA_ERR_NOENTITY);
}
static void command_load_module(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_module *m;
const char *name, *argument;
pa_tagstruct *reply;
- assert(c && t);
+ connection_assert_ref(c);
+ pa_assert(t);
+
if (pa_tagstruct_gets(t, &name) < 0 ||
pa_tagstruct_gets(t, &argument) < 0 ||
!pa_tagstruct_eof(t)) {
}
static void command_unload_module(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx;
pa_module *m;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
!pa_tagstruct_eof(t)) {
}
static void command_add_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const char *name, *module, *argument;
uint32_t type;
uint32_t idx;
pa_tagstruct *reply;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_gets(t, &name) < 0 ||
pa_tagstruct_getu32(t, &type) < 0 ||
}
static void command_remove_autoload(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const char *name = NULL;
uint32_t type, idx = PA_IDXSET_INVALID;
int r;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if ((pa_tagstruct_getu32(t, &idx) < 0 &&
(pa_tagstruct_gets(t, &name) < 0 ||
}
static void autoload_fill_tagstruct(pa_tagstruct *t, const pa_autoload_entry *e) {
- assert(t && e);
+ pa_assert(t && e);
pa_tagstruct_putu32(t, e->index);
pa_tagstruct_puts(t, e->name);
}
static void command_get_autoload_info(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
const pa_autoload_entry *a = NULL;
uint32_t type, idx;
const char *name;
pa_tagstruct *reply;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if ((pa_tagstruct_getu32(t, &idx) < 0 &&
(pa_tagstruct_gets(t, &name) < 0 ||
}
static void command_get_autoload_info_list(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
pa_tagstruct *reply;
- assert(c && t);
+
+ connection_assert_ref(c);
+ pa_assert(t);
if (!pa_tagstruct_eof(t)) {
protocol_error(c);
}
static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
- struct connection *c = userdata;
+ connection *c = CONNECTION(userdata);
uint32_t idx = PA_INVALID_INDEX, idx_device = PA_INVALID_INDEX;
const char *name = NULL;
- assert(c);
- assert(t);
+ connection_assert_ref(c);
+ pa_assert(t);
if (pa_tagstruct_getu32(t, &idx) < 0 ||
pa_tagstruct_getu32(t, &idx_device) < 0 ||
}
pa_pstream_send_simple_ack(c->pstream, tag);
-
}
/*** pstream callbacks ***/
static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_creds *creds, void *userdata) {
- struct connection *c = userdata;
- assert(p && packet && packet->data && c);
+ connection *c = CONNECTION(userdata);
+
+ pa_assert(p);
+ pa_assert(packet);
+ connection_assert_ref(c);
if (pa_pdispatch_run(c->pdispatch, packet, creds, c) < 0) {
pa_log("invalid packet.");
- connection_free(c);
+ connection_unlink(c);
}
}
static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata) {
- struct connection *c = userdata;
- struct output_stream *stream;
- assert(p && chunk && userdata);
-
- if (!(stream = pa_idxset_get_by_index(c->output_streams, channel))) {
+ connection *c = CONNECTION(userdata);
+ output_stream *stream;
+
+ pa_assert(p);
+ pa_assert(chunk);
+ connection_assert_ref(c);
+
+ if (!(stream = OUTPUT_STREAM(pa_idxset_get_by_index(c->output_streams, channel)))) {
pa_log("client sent block for invalid stream.");
/* Ignoring */
return;
}
- if (stream->type == PLAYBACK_STREAM) {
- struct playback_stream *ps = (struct playback_stream*) stream;
- if (chunk->length >= ps->requested_bytes)
- ps->requested_bytes = 0;
- else
- ps->requested_bytes -= chunk->length;
-
- pa_memblockq_seek(ps->memblockq, offset, seek);
-
- if (pa_memblockq_push_align(ps->memblockq, chunk) < 0) {
- pa_tagstruct *t;
-
- pa_log_warn("failed to push data into queue");
-
- /* Pushing this block into the queue failed, so we simulate
- * it by skipping ahead */
-
- pa_memblockq_seek(ps->memblockq, chunk->length, PA_SEEK_RELATIVE);
-
- /* Notify the user */
- t = pa_tagstruct_new(NULL, 0);
- pa_tagstruct_putu32(t, PA_COMMAND_OVERFLOW);
- pa_tagstruct_putu32(t, (uint32_t) -1); /* tag */
- pa_tagstruct_putu32(t, ps->index);
- pa_pstream_send_tagstruct(p, t);
- }
-
- ps->underrun = 0;
-
- pa_sink_notify(ps->sink_input->sink);
-
+ if (playback_stream_isinstance(stream)) {
+ playback_stream *ps = PLAYBACK_STREAM(stream);
+
+ if (seek != PA_SEEK_RELATIVE || offset != 0)
+ pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_SEEK, PA_UINT_TO_PTR(seek), offset, NULL, NULL);
+
+ pa_asyncmsgq_post(ps->sink_input->sink->asyncmsgq, PA_MSGOBJECT(ps->sink_input), SINK_INPUT_MESSAGE_POST_DATA, NULL, 0, chunk, NULL);
+
} else {
- struct upload_stream *u = (struct upload_stream*) stream;
+ upload_stream *u = UPLOAD_STREAM(stream);
size_t l;
- assert(u->type == UPLOAD_STREAM);
-
if (!u->memchunk.memblock) {
if (u->length == chunk->length) {
u->memchunk = *chunk;
}
}
- assert(u->memchunk.memblock);
+ pa_assert(u->memchunk.memblock);
l = u->length;
if (l > chunk->length)
}
static void pstream_die_callback(pa_pstream *p, void *userdata) {
- struct connection *c = userdata;
- assert(p && c);
- connection_free(c);
+ connection *c = CONNECTION(userdata);
+
+ pa_assert(p);
+ connection_assert_ref(c);
+ connection_unlink(c);
/* pa_log("connection died.");*/
}
static void pstream_drain_callback(pa_pstream *p, void *userdata) {
- struct connection *c = userdata;
- assert(p && c);
+ connection *c = CONNECTION(userdata);
+
+ pa_assert(p);
+ connection_assert_ref(c);
send_memblock(c);
}
/*** client callbacks ***/
static void client_kill_cb(pa_client *c) {
- assert(c && c->userdata);
- connection_free(c->userdata);
+ pa_assert(c);
+
+ connection_unlink(CONNECTION(c->userdata));
}
/*** socket server callbacks ***/
static void auth_timeout(pa_mainloop_api*m, pa_time_event *e, const struct timeval *tv, void *userdata) {
- struct connection *c = userdata;
- assert(m && tv && c && c->auth_timeout_event == e);
+ connection *c = CONNECTION(userdata);
+
+ pa_assert(m);
+ pa_assert(tv);
+ connection_assert_ref(c);
+ pa_assert(c->auth_timeout_event == e);
if (!c->authorized)
- connection_free(c);
+ connection_unlink(c);
}
static void on_connection(PA_GCC_UNUSED pa_socket_server*s, pa_iochannel *io, void *userdata) {
pa_protocol_native *p = userdata;
- struct connection *c;
+ connection *c;
char cname[256], pname[128];
- assert(io && p);
+
+ pa_assert(io);
+ pa_assert(p);
if (pa_idxset_size(p->connections)+1 > MAX_CONNECTIONS) {
pa_log_warn("Warning! Too many connections (%u), dropping incoming connection.", MAX_CONNECTIONS);
return;
}
- c = pa_xmalloc(sizeof(struct connection));
+ c = pa_msgobject_new(connection);
+ c->parent.parent.free = connection_free;
c->authorized = !!p->public;
c->protocol = p;
pa_iochannel_socket_peer_to_string(io, pname, sizeof(pname));
pa_snprintf(cname, sizeof(cname), "Native client (%s)", pname);
- assert(p->core);
+ pa_assert(p->core);
c->client = pa_client_new(p->core, __FILE__, cname);
- assert(c->client);
+ pa_assert(c->client);
c->client->kill = client_kill_cb;
c->client->userdata = c;
c->client->owner = p->module;
c->pstream = pa_pstream_new(p->core->mainloop, io, p->core->mempool);
- assert(c->pstream);
+ pa_assert(c->pstream);
pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
c->pdispatch = pa_pdispatch_new(p->core->mainloop, command_table, PA_COMMAND_MAX);
- assert(c->pdispatch);
+ pa_assert(c->pdispatch);
c->record_streams = pa_idxset_new(NULL, NULL);
c->output_streams = pa_idxset_new(NULL, NULL);
- assert(c->record_streams && c->output_streams);
+ pa_assert(c->record_streams && c->output_streams);
c->rrobin_index = PA_IDXSET_INVALID;
c->subscription = NULL;
/*** module entry points ***/
static int load_key(pa_protocol_native*p, const char*fn) {
- assert(p);
+ pa_assert(p);
p->auth_cookie_in_property = 0;
int public = 0;
const char *acl;
- assert(c);
- assert(ma);
+ pa_assert(c);
+ pa_assert(ma);
if (pa_modargs_get_value_boolean(ma, "auth-anonymous", &public) < 0) {
pa_log("auth-anonymous= expects a boolean argument.");
goto fail;
p->connections = pa_idxset_new(NULL, NULL);
- assert(p->connections);
+ pa_assert(p->connections);
return p;
}
void pa_protocol_native_free(pa_protocol_native *p) {
- struct connection *c;
- assert(p);
+ connection *c;
+ pa_assert(p);
while ((c = pa_idxset_first(p->connections, NULL)))
- connection_free(c);
+ connection_unlink(c);
pa_idxset_free(p->connections, NULL, NULL);
if (p->server) {
pa_xfree(p);
}
-pa_protocol_native* pa_protocol_native_new_iochannel(pa_core*core, pa_iochannel *io, pa_module *m, pa_modargs *ma) {
+pa_protocol_native* pa_protocol_native_new_iochannel(
+ pa_core*core,
+ pa_iochannel *io,
+ pa_module *m,
+ pa_modargs *ma) {
+
pa_protocol_native *p;
if (!(p = protocol_new_internal(core, m, ma)))