From eecf602476ff5b51bdc08f8fd0e4aa70d2b0ef5a Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Sun, 20 Jun 2004 01:12:13 +0000 Subject: partial implementation of native protocol git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@30 fefdeb5f-60dc-0310-8127-8f9354f1896f --- src/Makefile.am | 27 +++- src/module-protocol-stub.c | 13 +- src/packet.c | 21 ++- src/packet.h | 6 +- src/protocol-native-tcp.c | 19 --- src/protocol-native-unix.c | 27 ---- src/protocol-native.c | 349 ++++++++++++++++++++++++++++++++++++++++++--- src/protocol-native.h | 5 +- src/protocol-simple.c | 1 + src/pstream.c | 95 +++++++----- src/pstream.h | 6 +- src/sample.h | 2 +- src/sinkinput.h | 1 - src/tagstruct.c | 187 ++++++++++++++++++++++++ src/tagstruct.h | 30 ++++ src/todo | 1 - 16 files changed, 680 insertions(+), 110 deletions(-) delete mode 100644 src/protocol-native-tcp.c delete mode 100644 src/protocol-native-unix.c create mode 100644 src/tagstruct.c create mode 100644 src/tagstruct.h (limited to 'src') diff --git a/src/Makefile.am b/src/Makefile.am index fc7b3933..443a25f2 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1,4 +1,4 @@ -# $Id: Makefile.am 27 2003-10-22 22:34:06Z lennart $ +# $Id$ # # This file is part of polypaudio. # @@ -25,7 +25,8 @@ pkglib_LTLIBRARIES=libprotocol-simple.la module-simple-protocol-tcp.la \ libpacket.la module-oss.la module-oss-mmap.la liboss.la libioline.la \ libcli.la module-cli.la libtokenizer.la libdynarray.la \ module-simple-protocol-unix.la module-cli-protocol-tcp.la \ - libprotocol-cli.la + libprotocol-cli.la libprotocol-native.la module-native-protocol-tcp.la \ + module-native-protocol-unix.la module-cli-protocol-unix.la libtagstruct.la polypaudio_SOURCES = idxset.c idxset.h \ queue.c queue.h \ @@ -87,6 +88,13 @@ libprotocol_cli_la_SOURCES = protocol-cli.c protocol-cli.h libprotocol_cli_la_LDFLAGS = -avoid-version libprotocol_cli_la_LIBADD = libsocket-server.la libiochannel.la libcli.la +libprotocol_native_la_SOURCES = protocol-native.c protocol-native.h +libprotocol_native_la_LDFLAGS = -avoid-version +libprotocol_native_la_LIBADD = libsocket-server.la libiochannel.la libpacket.la libpstream.la + +libtagstruct_la_SOURCES = tagstruct.c tagstruct.h +libtagstruct_la_LDFLAGS = -avoid-version + module_simple_protocol_tcp_la_SOURCES = module-protocol-stub.c module_simple_protocol_tcp_la_CFLAGS = -DUSE_TCP_SOCKETS -DUSE_PROTOCOL_SIMPLE $(AM_CFLAGS) module_simple_protocol_tcp_la_LDFLAGS = -module -avoid-version @@ -102,6 +110,21 @@ module_cli_protocol_tcp_la_CFLAGS = -DUSE_TCP_SOCKETS -DUSE_PROTOCOL_CLI $(AM_CF module_cli_protocol_tcp_la_LDFLAGS = -module -avoid-version module_cli_protocol_tcp_la_LIBADD = libprotocol-cli.la libiochannel.la +module_cli_protocol_unix_la_SOURCES = module-protocol-stub.c +module_cli_protocol_unix_la_CFLAGS = -DUSE_UNIX_SOCKETS -DUSE_PROTOCOL_CLI $(AM_CFLAGS) +module_cli_protocol_unix_la_LDFLAGS = -module -avoid-version +module_cli_protocol_unix_la_LIBADD = libprotocol-cli.la libiochannel.la + +module_native_protocol_tcp_la_SOURCES = module-protocol-stub.c +module_native_protocol_tcp_la_CFLAGS = -DUSE_TCP_SOCKETS -DUSE_PROTOCOL_NATIVE $(AM_CFLAGS) +module_native_protocol_tcp_la_LDFLAGS = -module -avoid-version +module_native_protocol_tcp_la_LIBADD = libprotocol-native.la libiochannel.la libtagstruct.la + +module_native_protocol_unix_la_SOURCES = module-protocol-stub.c +module_native_protocol_unix_la_CFLAGS = -DUSE_UNIX_SOCKETS -DUSE_PROTOCOL_NATIVE $(AM_CFLAGS) +module_native_protocol_unix_la_LDFLAGS = -module -avoid-version +module_native_protocol_unix_la_LIBADD = libprotocol-native.la libiochannel.la libtagstruct.la + module_pipe_sink_la_SOURCES = module-pipe-sink.c module_pipe_sink_la_LDFLAGS = -module -avoid-version module_pipe_sink_la_LIBADD = libiochannel.la diff --git a/src/module-protocol-stub.c b/src/module-protocol-stub.c index 9cbf236e..2387017c 100644 --- a/src/module-protocol-stub.c +++ b/src/module-protocol-stub.c @@ -7,15 +7,22 @@ #ifdef USE_PROTOCOL_SIMPLE #include "protocol-simple.h" #define protocol_free protocol_simple_free - #define IPV4_PORT 4712 + #define IPV4_PORT 4711 #else #ifdef USE_PROTOCOL_CLI #include "protocol-cli.h" #define protocol_new protocol_cli_new #define protocol_free protocol_cli_free - #define IPV4_PORT 4711 + #define IPV4_PORT 4712 #else - #error "Broken build system" + #ifdef USE_PROTOCOL_NATIVE + #include "protocol-native.h" + #define protocol_new protocol_native_new + #define protocol_free protocol_native_free + #define IPV4_PORT 4713 + #else + #error "Broken build system" + #endif #endif #endif diff --git a/src/packet.c b/src/packet.c index 086e4b2a..47fce919 100644 --- a/src/packet.c +++ b/src/packet.c @@ -3,7 +3,7 @@ #include "packet.h" -struct packet* packet_new(uint32_t length) { +struct packet* packet_new(size_t length) { struct packet *p; assert(length); p = malloc(sizeof(struct packet)+length); @@ -11,9 +11,23 @@ struct packet* packet_new(uint32_t length) { p->ref = 1; p->length = length; + p->data = (uint8_t*) (p+1); + p->type = PACKET_APPENDED; return p; } +struct packet* packet_dynamic(uint8_t* data, size_t length) { + struct packet *p; + assert(data && length); + p = malloc(sizeof(struct packet)); + assert(p); + + p->ref = 1; + p->length = length; + p->data = data; + p->type = PACKET_DYNAMIC; +} + struct packet* packet_ref(struct packet *p) { assert(p && p->ref >= 1); p->ref++; @@ -24,6 +38,9 @@ void packet_unref(struct packet *p) { assert(p && p->ref >= 1); p->ref--; - if (p->ref == 0) + if (p->ref == 0) { + if (p->type == PACKET_DYNAMIC) + free(p->data); free(p); + } } diff --git a/src/packet.h b/src/packet.h index 781c0e66..7e2e0066 100644 --- a/src/packet.h +++ b/src/packet.h @@ -5,12 +5,14 @@ #include struct packet { + enum { PACKET_APPENDED, PACKET_DYNAMIC } type; unsigned ref; size_t length; - uint8_t data[]; + uint8_t *data; }; -struct packet* packet_new(uint32_t length); +struct packet* packet_new(size_t length); +struct packet* packet_new_dynamic(uint8_t* data, size_t length); struct packet* packet_ref(struct packet *p); void packet_unref(struct packet *p); diff --git a/src/protocol-native-tcp.c b/src/protocol-native-tcp.c deleted file mode 100644 index b33f3e15..00000000 --- a/src/protocol-native-tcp.c +++ /dev/null @@ -1,19 +0,0 @@ -#include "module.h" - -int module_init(struct core *c, struct module*m) { - struct socket_server *s; - assert(c && m); - - if (!(s = socket_server_new_ipv4(c->mainloop, INADDR_LOOPBACK, 4711))) - return -1; - - m->userdata = protocol_native_new(s); - assert(m->userdata); - return 0; -} - -void module_done(struct core *c, struct module*m) { - assert(c && m); - - protocol_native_free(m->userdata); -} diff --git a/src/protocol-native-unix.c b/src/protocol-native-unix.c deleted file mode 100644 index a18965cd..00000000 --- a/src/protocol-native-unix.c +++ /dev/null @@ -1,27 +0,0 @@ -#include "module.h" - -int module_init(struct core *c, struct module*m) { - struct fn[PATH_MAX]; - struct socket_server *s; - char *t; - assert(c && m); - - if (!(t = getenv("TMP"))) - if (!(t = getenv("TEMP"))) - t = "/tmp"; - - snprintf(fn, sizeof(fn), "%s/foosock", t); - - if (!(s = socket_server_new_unix(c->mainloop, fn))) - return -1; - - m->userdata = protocol_native_new(s); - assert(m->userdata); - return 0; -} - -void module_done(struct core *c, struct module*m) { - assert(c && m); - - protocol_native_free(m->userdata); -} diff --git a/src/protocol-native.c b/src/protocol-native.c index bdb69355..e9cca7c1 100644 --- a/src/protocol-native.c +++ b/src/protocol-native.c @@ -1,49 +1,364 @@ +#include +#include +#include + #include "protocol-native.h" +#include "packet.h" +#include "client.h" +#include "sourceoutput.h" +#include "sinkinput.h" +#include "pstream.h" +#include "tagstruct.h" -struct protocol_native { - struct socket_server*server; - struct idxset *connection; +struct connection; +struct protocol_native; + +enum { + COMMAND_ERROR, + COMMAND_REPLY, + COMMAND_CREATE_PLAYBACK_STREAM, + COMMAND_DELETE_PLAYBACK_STREAM, + COMMAND_CREATE_RECORD_STREAM, + COMMAND_DELETE_RECORD_STREAM, + COMMAND_EXIT, + COMMAND_MAX }; -struct stream_info { - guint32_t tag; - - union { - struct output_stream *output_stream; - struct input_stream *input_stream; - } +enum { + ERROR_ACCESS, + ERROR_COMMAND, + ERROR_ARGUMENT, + ERROR_EXIST +}; + +struct record_stream { + struct connection *connection; + uint32_t index; + struct source_output *source_output; + struct memblockq *memblockq; +}; + +struct playback_stream { + struct connection *connection; + uint32_t index; + struct sink_input *sink_input; + struct memblockq *memblockq; }; struct connection { + int authorized; + struct protocol_native *protocol; struct client *client; - struct serializer *serializer; + struct pstream *pstream; + struct idxset *record_streams, *playback_streams; +}; +struct protocol_native { + int public; + struct core *core; + struct socket_server *server; + struct idxset *connections; +}; + +static void record_stream_free(struct record_stream* r) { + assert(r && r->connection); + + idxset_remove_by_data(r->connection->record_streams, r, NULL); + source_output_free(r->source_output); + memblockq_free(r->memblockq); + free(r); +} + +static struct playback_stream* playback_stream_new(struct connection *c, struct sink *sink, struct sample_spec *ss, const char *name, size_t maxlength, size_t prebuf) { + struct playback_stream *s; + + s = malloc(sizeof(struct playback_stream)); + assert (s); + s->connection = c; + s->sink_input = sink_input_new(sink, ss, name); + assert(s->sink_input); + s->memblockq = memblockq_new(maxlength, sample_size(ss), prebuf); + assert(s->memblockq); + + idxset_put(c->playback_streams, s, &s->index); + return s; +} + +static void playback_stream_free(struct playback_stream* p) { + assert(p && p->connection); + + idxset_remove_by_data(p->connection->playback_streams, p, NULL); + sink_input_free(p->sink_input); + memblockq_free(p->memblockq); + free(p); +} + +static void connection_free(struct connection *c) { + struct record_stream *r; + struct playback_stream *p; + assert(c && c->protocol); + + idxset_remove_by_data(c->protocol->connections, c, NULL); + pstream_free(c->pstream); + while ((r = idxset_first(c->record_streams, NULL))) + record_stream_free(r); + idxset_free(c->record_streams, NULL, NULL); + + while ((p = idxset_first(c->playback_streams, NULL))) + playback_stream_free(p); + idxset_free(c->playback_streams, NULL, NULL); + + client_free(c->client); + free(c); +} + +/*** pstream callbacks ***/ + +static void send_tagstruct(struct pstream *p, struct tagstruct *t) { + size_t length; + uint8_t *data; + struct packet *packet; + assert(p && t); + + data = tagstruct_free_data(t, &length); + assert(data && length); + packet = packet_new_dynamic(data, length); + assert(packet); + pstream_send_packet(p, packet); + packet_unref(packet); +} + +static void send_error(struct pstream *p, uint32_t tag, uint32_t error) { + struct tagstruct *t = tagstruct_new(NULL, 0); + assert(t); + tagstruct_putu32(t, COMMAND_ERROR); + tagstruct_putu32(t, tag); + tagstruct_putu32(t, error); + send_tagstruct(p, t); +} + +static void send_simple_ack(struct pstream *p, uint32_t tag) { + struct tagstruct *t = tagstruct_new(NULL, 0); + assert(t); + tagstruct_putu32(t, COMMAND_REPLY); + tagstruct_putu32(t, tag); + send_tagstruct(p, t); +} + +struct command { + int (*func)(struct connection *c, uint32_t tag, struct tagstruct *t); +}; + +static int command_create_playback_stream(struct connection *c, uint32_t tag, struct tagstruct *t) { + struct playback_stream *s; + size_t maxlength, prebuf; + uint32_t sink_index; + const char *name; + struct sample_spec ss; + struct tagstruct *reply; + struct sink *sink; + assert(c && t && c->protocol && c->protocol->core); + + if (tagstruct_gets(t, &name) < 0 || + tagstruct_get_sample_spec(t, &ss) < 0 || + tagstruct_getu32(t, &sink_index) < 0 || + tagstruct_getu32(t, &maxlength) < 0 || + tagstruct_getu32(t, &prebuf) < 0 || + !tagstruct_eof(t)) + return -1; + + if (!c->authorized) { + send_error(c->pstream, tag, ERROR_ACCESS); + return 0; + } + + if (sink_index == (uint32_t) -1) + sink = sink_get_default(c->protocol->core); + else + sink = idxset_get_by_index(c->protocol->core->sinks, sink_index); + + if (!sink) { + send_error(c->pstream, tag, ERROR_EXIST); + return 0; + } + if (!(s = playback_stream_new(c, sink, &ss, name, maxlength, prebuf))) { + send_error(c->pstream, tag, ERROR_ARGUMENT); + return 0; + } + + reply = tagstruct_new(NULL, 0); + assert(reply); + tagstruct_putu32(reply, COMMAND_REPLY); + tagstruct_putu32(reply, tag); + tagstruct_putu32(reply, s->index); + send_tagstruct(c->pstream, reply); + return 0; +} + +static int command_delete_playback_stream(struct connection *c, uint32_t tag, struct tagstruct *t) { + uint32_t channel; + struct playback_stream *s; + assert(c && t); + + if (tagstruct_getu32(t, &channel) < 0 || + !tagstruct_eof(t)) + return -1; + + if (!c->authorized) { + send_error(c->pstream, tag, ERROR_ACCESS); + return 0; + } + + if (!(s = idxset_get_by_index(c->playback_streams, channel))) { + send_error(c->pstream, tag, ERROR_EXIST); + return 0; + } + + send_simple_ack(c->pstream, tag); + return 0; +} + +static int command_exit(struct connection *c, uint32_t tag, struct tagstruct *t) { + assert(c && t); + + if (!tagstruct_eof(t)) + return -1; + + if (!c->authorized) { + send_error(c->pstream, tag, ERROR_ACCESS); + return 0; + } + + assert(c->protocol && c->protocol->core); + mainloop_quit(c->protocol->core->mainloop, -1); + send_simple_ack(c->pstream, tag); /* nonsense */ + return 0; +} + +static const struct command commands[] = { + [COMMAND_ERROR] = { NULL }, + [COMMAND_REPLY] = { NULL }, + [COMMAND_CREATE_PLAYBACK_STREAM] = { command_create_playback_stream }, + [COMMAND_DELETE_PLAYBACK_STREAM] = { command_delete_playback_stream }, + [COMMAND_CREATE_RECORD_STREAM] = { NULL }, + [COMMAND_DELETE_RECORD_STREAM] = { NULL }, + [COMMAND_EXIT] = { command_exit }, }; -static void on_connection(struct socket_server *server, struct iochannel *io, void *userdata) { - struct protocol_native *p = userdata; - assert(server && io && p && p->server == server); +static int packet_callback(struct pstream *p, struct packet *packet, void *userdata) { + struct connection *c = userdata; + uint32_t tag, command; + struct tagstruct *ts = NULL; + assert(p && packet && packet->data && c); + + if (packet->length <= 8) + goto fail; + + ts = tagstruct_new(packet->data, packet->length); + assert(ts); + + if (tagstruct_getu32(ts, &command) < 0 || + tagstruct_getu32(ts, &tag) < 0) + goto fail; + if (command >= COMMAND_MAX || !commands[command].func) + send_error(p, tag, ERROR_COMMAND); + else if (commands[command].func(c, tag, ts) < 0) + goto fail; + tagstruct_free(ts); + + return 0; + +fail: + if (ts) + tagstruct_free(ts); + + fprintf(stderr, "protocol-native: invalid packet.\n"); + return -1; + +} + +static int memblock_callback(struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata) { + struct connection *c = userdata; + struct playback_stream *stream; + assert(p && chunk && userdata); + + if (!(stream = idxset_get_by_index(c->playback_streams, channel))) { + fprintf(stderr, "protocol-native: client sent block for invalid stream.\n"); + return -1; + } + + memblockq_push(stream->memblockq, chunk, delta); + assert(stream->sink_input); + sink_notify(stream->sink_input->sink); + + return 0; +} + +static void die_callback(struct pstream *p, void *userdata) { + struct connection *c = userdata; + assert(p && c); + connection_free(c); + + fprintf(stderr, "protocol-native: connection died.\n"); } -struct protocol_native* protocol_native(struct socket_server *server) { +/*** socket server callbacks ***/ + +static void on_connection(struct socket_server*s, struct iochannel *io, void *userdata) { + struct protocol_native *p = userdata; + struct connection *c; + assert(s && io && p); + + c = malloc(sizeof(struct connection)); + assert(c); + c->authorized = p->public; + c->protocol = p; + assert(p->core); + c->client = client_new(p->core, "NATIVE", "Client"); + assert(c->client); + c->pstream = pstream_new(p->core->mainloop, io); + assert(c->pstream); + + pstream_set_recieve_packet_callback(c->pstream, packet_callback, c); + pstream_set_recieve_memblock_callback(c->pstream, memblock_callback, c); + pstream_set_die_callback(c->pstream, die_callback, c); + + c->record_streams = idxset_new(NULL, NULL); + c->playback_streams = idxset_new(NULL, NULL); + assert(c->record_streams && c->playback_streams); + + idxset_put(p->connections, c, NULL); +} + +/*** module entry points ***/ + +struct protocol_native* protocol_native_new(struct core *core, struct socket_server *server) { struct protocol_native *p; - assert(server); + assert(core && server); p = malloc(sizeof(struct protocol_native)); assert(p); + p->public = 1; p->server = server; - socket_server_set_callback(p->server, callback, p); + p->core = core; + p->connections = idxset_new(NULL, NULL); + socket_server_set_callback(p->server, on_connection, p); + return p; } void protocol_native_free(struct protocol_native *p) { + struct connection *c; assert(p); + while ((c = idxset_first(p->connections, NULL))) + connection_free(c); + idxset_free(p->connections, NULL, NULL); socket_server_free(p->server); free(p); } diff --git a/src/protocol-native.h b/src/protocol-native.h index bdad03b4..88283e1c 100644 --- a/src/protocol-native.h +++ b/src/protocol-native.h @@ -1,9 +1,12 @@ #ifndef fooprotocolnativehfoo #define fooprotocolnativehfoo +#include "core.h" +#include "socket-server.h" + struct protocol_native; -struct protocol_native* protocol_native(struct socket_server *server); +struct protocol_native* protocol_native_new(struct core*core, struct socket_server *server); void protocol_native_free(struct protocol_native *n); #endif diff --git a/src/protocol-simple.c b/src/protocol-simple.c index a0a996cb..8e4246cd 100644 --- a/src/protocol-simple.c +++ b/src/protocol-simple.c @@ -79,6 +79,7 @@ static int do_read(struct connection *c) { assert(c->input_memblockq); memblockq_push(c->input_memblockq, &chunk, 0); memblock_unref(chunk.memblock); + assert(c->sink_input); sink_notify(c->sink_input->sink); return 0; diff --git a/src/pstream.c b/src/pstream.c index 0336d161..a63e126d 100644 --- a/src/pstream.c +++ b/src/pstream.c @@ -1,5 +1,6 @@ #include #include +#include #include "pstream.h" #include "queue.h" @@ -35,6 +36,8 @@ struct pstream { struct queue *send_queue; int dead; + void (*die_callback) (struct pstream *p, void *userdad); + void *die_callback_userdata; struct { struct item_info* current; @@ -54,10 +57,10 @@ struct pstream { size_t index; } read; - void (*recieve_packet_callback) (struct pstream *p, struct packet *packet, void *userdata); + int (*recieve_packet_callback) (struct pstream *p, struct packet *packet, void *userdata); void *recieve_packet_callback_userdata; - void (*recieve_memblock_callback) (struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata); + int (*recieve_memblock_callback) (struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata); void *recieve_memblock_callback_userdata; }; @@ -89,6 +92,8 @@ struct pstream *pstream_new(struct mainloop *m, struct iochannel *io) { iochannel_set_callback(io, io_callback, p); p->dead = 0; + p->die_callback = NULL; + p->die_callback_userdata = NULL; p->mainloop = m; p->mainloop_source = mainloop_source_new_fixed(m, prepare_callback, p); @@ -165,7 +170,7 @@ void pstream_send_packet(struct pstream*p, struct packet *packet) { i = malloc(sizeof(struct item_info)); assert(i); i->type = PSTREAM_ITEM_PACKET; - i->packet = packet; + i->packet = packet_ref(packet); queue_push(p->send_queue, i); mainloop_source_enable(p->mainloop_source, 1); @@ -182,18 +187,20 @@ void pstream_send_memblock(struct pstream*p, uint32_t channel, int32_t delta, st i->channel = channel; i->delta = delta; + memblock_ref(i->chunk.memblock); + queue_push(p->send_queue, i); mainloop_source_enable(p->mainloop_source, 1); } -void pstream_set_recieve_packet_callback(struct pstream *p, void (*callback) (struct pstream *p, struct packet *packet, void *userdata), void *userdata) { +void pstream_set_recieve_packet_callback(struct pstream *p, int (*callback) (struct pstream *p, struct packet *packet, void *userdata), void *userdata) { assert(p && callback); p->recieve_packet_callback = callback; p->recieve_packet_callback_userdata = userdata; } -void pstream_set_recieve_memblock_callback(struct pstream *p, void (*callback) (struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata), void *userdata) { +void pstream_set_recieve_memblock_callback(struct pstream *p, int (*callback) (struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata), void *userdata) { assert(p && callback); p->recieve_memblock_callback = callback; @@ -211,15 +218,15 @@ static void prepare_next_write_item(struct pstream *p) { if (p->write.current->type == PSTREAM_ITEM_PACKET) { assert(p->write.current->packet); p->write.data = p->write.current->packet->data; - p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH] = p->write.current->packet->length; + p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length); p->write.descriptor[PSTREAM_DESCRIPTOR_CHANNEL] = 0; p->write.descriptor[PSTREAM_DESCRIPTOR_DELTA] = 0; } else { assert(p->write.current->type == PSTREAM_ITEM_MEMBLOCK && p->write.current->chunk.memblock); p->write.data = p->write.current->chunk.memblock->data + p->write.current->chunk.index; - p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH] = p->write.current->chunk.length; - p->write.descriptor[PSTREAM_DESCRIPTOR_CHANNEL] = p->write.current->channel; - p->write.descriptor[PSTREAM_DESCRIPTOR_DELTA] = p->write.current->delta; + p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length); + p->write.descriptor[PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel); + p->write.descriptor[PSTREAM_DESCRIPTOR_DELTA] = htonl(p->write.current->delta); } } @@ -247,17 +254,15 @@ static void do_write(struct pstream *p) { l = PSTREAM_DESCRIPTOR_SIZE - p->write.index; } else { d = (void*) p->write.data + p->write.index - PSTREAM_DESCRIPTOR_SIZE; - l = p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH] - p->write.index - PSTREAM_DESCRIPTOR_SIZE; + l = ntohl(p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH]) - p->write.index - PSTREAM_DESCRIPTOR_SIZE; } - if ((r = iochannel_write(p->io, d, l)) < 0) { - p->dead = 1; - return; - } + if ((r = iochannel_write(p->io, d, l)) < 0) + goto die; p->write.index += r; - if (p->write.index >= PSTREAM_DESCRIPTOR_SIZE+p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH]) { + if (p->write.index >= PSTREAM_DESCRIPTOR_SIZE+ntohl(p->write.descriptor[PSTREAM_DESCRIPTOR_LENGTH])) { assert(p->write.current); item_free(p->write.current, (void *) 1); p->write.current = NULL; @@ -265,6 +270,13 @@ static void do_write(struct pstream *p) { if (p->send_callback && queue_is_empty(p->send_queue)) p->send_callback(p, p->send_callback_userdata); } + + return; + +die: + p->dead = 1; + if (p->die_callback) + p->die_callback(p, p->die_callback_userdata); } static void do_read(struct pstream *p) { @@ -284,35 +296,31 @@ static void do_read(struct pstream *p) { } else { assert(p->read.data); d = (void*) p->read.data + p->read.index - PSTREAM_DESCRIPTOR_SIZE; - l = p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH] - p->read.index - PSTREAM_DESCRIPTOR_SIZE; - } - - if ((r = iochannel_read(p->io, d, l)) <= 0) { - p->dead = 1; - return; + l = ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH]) - p->read.index - PSTREAM_DESCRIPTOR_SIZE; } + if ((r = iochannel_read(p->io, d, l)) <= 0) + goto die; + p->read.index += r; if (p->read.index == PSTREAM_DESCRIPTOR_SIZE) { /* Reading of frame descriptor complete */ /* Frame size too large */ - if (p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH] > FRAME_SIZE_MAX) { - p->dead = 1; - return; - } + if (ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH]) > FRAME_SIZE_MAX) + goto die; assert(!p->read.packet && !p->read.memblock); - if (p->read.descriptor[PSTREAM_DESCRIPTOR_CHANNEL] == 0) { + if (ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_CHANNEL]) == 0) { /* Frame is a packet frame */ - p->read.packet = packet_new(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH]); + p->read.packet = packet_new(ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH])); assert(p->read.packet); p->read.data = p->read.packet->data; } else { /* Frame is a memblock frame */ - p->read.memblock = memblock_new(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH]); + p->read.memblock = memblock_new(ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH])); assert(p->read.memblock); p->read.data = p->read.memblock->data; } @@ -320,7 +328,7 @@ static void do_read(struct pstream *p) { } else if (p->read.index > PSTREAM_DESCRIPTOR_SIZE) { /* Frame payload available */ - if (p->read.memblock && p->recieve_memblock_callback) { /* Is this memblockd data? Than pass it to the user */ + if (p->read.memblock && p->recieve_memblock_callback) { /* Is this memblock data? Than pass it to the user */ size_t l; l = p->read.index - r < PSTREAM_DESCRIPTOR_SIZE ? p->read.index - PSTREAM_DESCRIPTOR_SIZE : r; @@ -332,28 +340,51 @@ static void do_read(struct pstream *p) { chunk.index = p->read.index - PSTREAM_DESCRIPTOR_SIZE - l; chunk.length = l; - p->recieve_memblock_callback(p, p->read.descriptor[PSTREAM_DESCRIPTOR_CHANNEL], (int32_t) p->read.descriptor[PSTREAM_DESCRIPTOR_DELTA], &chunk, p->recieve_memblock_callback_userdata); + if (p->recieve_memblock_callback(p, + ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_CHANNEL]), + (int32_t) ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_DELTA]), + &chunk, + p->recieve_memblock_callback_userdata) < 0) + goto die; } } /* Frame complete */ - if (p->read.index >= p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH] + PSTREAM_DESCRIPTOR_SIZE) { + if (p->read.index >= ntohl(p->read.descriptor[PSTREAM_DESCRIPTOR_LENGTH]) + PSTREAM_DESCRIPTOR_SIZE) { if (p->read.memblock) { assert(!p->read.packet); memblock_unref(p->read.memblock); p->read.memblock = NULL; } else { + int r = 0; assert(p->read.packet); if (p->recieve_packet_callback) - p->recieve_packet_callback(p, p->read.packet, p->recieve_packet_callback_userdata); + r = p->recieve_packet_callback(p, p->read.packet, p->recieve_packet_callback_userdata); packet_unref(p->read.packet); p->read.packet = NULL; + + if (r < 0) + goto die; } p->read.index = 0; } } + + return; + +die: + p->dead = 1; + if (p->die_callback) + p->die_callback(p, p->die_callback_userdata); + +} + +void pstream_set_die_callback(struct pstream *p, void (*callback)(struct pstream *p, void *userdata), void *userdata) { + assert(p && callback); + p->die_callback = callback; + p->die_callback_userdata = userdata; } diff --git a/src/pstream.h b/src/pstream.h index c0b57496..7113681e 100644 --- a/src/pstream.h +++ b/src/pstream.h @@ -16,7 +16,9 @@ void pstream_set_send_callback(struct pstream*p, void (*callback) (struct pstrea void pstream_send_packet(struct pstream*p, struct packet *packet); void pstream_send_memblock(struct pstream*p, uint32_t channel, int32_t delta, struct memchunk *chunk); -void pstream_set_recieve_packet_callback(struct pstream *p, void (*callback) (struct pstream *p, struct packet *packet, void *userdata), void *userdata); -void pstream_set_recieve_memblock_callback(struct pstream *p, void (*callback) (struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata), void *userdata); +void pstream_set_recieve_packet_callback(struct pstream *p, int (*callback) (struct pstream *p, struct packet *packet, void *userdata), void *userdata); +void pstream_set_recieve_memblock_callback(struct pstream *p, int (*callback) (struct pstream *p, uint32_t channel, int32_t delta, struct memchunk *chunk, void *userdata), void *userdata); + +void pstream_set_die_callback(struct pstream *p, void (*callback)(struct pstream *p, void *userdata), void *userdata); #endif diff --git a/src/sample.h b/src/sample.h index 5ed740e4..b2f13cc4 100644 --- a/src/sample.h +++ b/src/sample.h @@ -19,7 +19,7 @@ enum sample_format { struct sample_spec { enum sample_format format; uint32_t rate; - uint32_t channels; + uint8_t channels; }; #define DEFAULT_SAMPLE_SPEC default_sample_spec diff --git a/src/sinkinput.h b/src/sinkinput.h index f62070c4..389d832d 100644 --- a/src/sinkinput.h +++ b/src/sinkinput.h @@ -19,7 +19,6 @@ struct sink_input { void (*drop) (struct sink_input *i, size_t length); void (*kill) (struct sink_input *i); uint32_t (*get_latency) (struct sink_input *i); - void *userdata; }; diff --git a/src/tagstruct.c b/src/tagstruct.c new file mode 100644 index 00000000..429dd408 --- /dev/null +++ b/src/tagstruct.c @@ -0,0 +1,187 @@ +#include +#include +#include +#include + +#include "tagstruct.h" + +enum tags { + TAG_STRING = 't', + TAG_U32 = 'L', + TAG_S32 = 'l', + TAG_U16 = 'S', + TAG_S16 = 's', + TAG_U8 = 'B', + TAG_S8 = 'b', + TAG_SAMPLE_SPEC = 'a' +}; + +struct tagstruct { + uint8_t *data; + size_t length, allocated; + size_t rindex; + + int dynamic; +}; + +struct tagstruct *tagstruct_new(const uint8_t* data, size_t length) { + struct tagstruct*t; + + assert(!data || (data && length)); + + t = malloc(sizeof(struct tagstruct)); + assert(t); + t->data = (uint8_t*) data; + t->allocated = t->length = data ? length : 0; + t->rindex = 0; + t->dynamic = !!data; + return t; +} + +void tagstruct_free(struct tagstruct*t) { + assert(t); + if (t->dynamic) + free(t->data); + free(t); +} + +uint8_t* tagstruct_free_data(struct tagstruct*t, size_t *l) { + uint8_t *p; + assert(t && t->dynamic && l); + p = t->data; + *l = t->length; + free(t); + return p; +} + +static void extend(struct tagstruct*t, size_t l) { + assert(t && t->dynamic); + + if (t->allocated <= l) + return; + + t->data = realloc(t->data, t->allocated = l+100); + assert(t->data); +} + +void tagstruct_puts(struct tagstruct*t, const char *s) { + size_t l; + assert(t && s); + l = strlen(s)+2; + extend(t, l); + t->data[t->length] = TAG_STRING; + strcpy(t->data+t->length+1, s); + t->length += l; +} + +void tagstruct_putu32(struct tagstruct*t, uint32_t i) { + assert(t && i); + extend(t, 5); + t->data[t->length] = TAG_U32; + *((uint32_t*) (t->data+t->length+1)) = htonl(i); + t->length += 5; +} + +void tagstruct_putu8(struct tagstruct*t, uint8_t c) { + assert(t && c); + extend(t, 2); + t->data[t->length] = TAG_U8; + *(t->data+t->length+1) = c; + t->length += 2; +} + +void tagstruct_put_sample_spec(struct tagstruct *t, struct sample_spec *ss) { + assert(t && ss); + extend(t, 7); + t->data[t->length] = TAG_SAMPLE_SPEC; + t->data[t->length+1] = (uint8_t) ss->format; + t->data[t->length+2] = ss->channels; + *(uint32_t*) (t->data+t->length+3) = htonl(ss->rate); + t->length += 7; +} + +int tagstruct_gets(struct tagstruct*t, const char **s) { + int error = 0; + size_t n; + char *c; + assert(t && s); + + if (t->rindex+2 > t->length) + return -1; + + if (t->data[t->rindex] != TAG_STRING) + return -1; + + error = 1; + for (n = 0, c = (char*) (t->data+t->rindex+1); n < t->length-t->rindex-1; c++) + if (!*c) { + error = 0; + break; + } + + if (error) + return -1; + + *s = (char*) (t->data+t->rindex+1); + + t->rindex += n+1; + return 0; +} + +int tagstruct_getu32(struct tagstruct*t, uint32_t *i) { + assert(t && i); + + if (t->rindex+5 > t->length) + return -1; + + if (t->data[t->rindex] != TAG_U32) + return -1; + + *i = ntohl(*((uint32_t*) (t->data+t->rindex+1))); + t->rindex += 5; + return 0; +} + +int tagstruct_getu8(struct tagstruct*t, uint8_t *c) { + assert(t && c); + + if (t->rindex+2 > t->length) + return -1; + + if (t->data[t->rindex] != TAG_U8) + return -1; + + *c = t->data[t->rindex+1]; + t->rindex +=2; + return 0; +} + +int tagstruct_get_sample_spec(struct tagstruct *t, struct sample_spec *ss) { + assert(t && ss); + + if (t->rindex+7 > t->length) + return -1; + + if (t->data[t->rindex] != TAG_SAMPLE_SPEC) + return -1; + + ss->format = t->data[t->rindex+1]; + ss->channels = t->data[t->rindex+2]; + ss->rate = ntohl(*(uint32_t*) (t->data+t->rindex+3)); + + t->rindex += 7; + return 0; +} + + +int tagstruct_eof(struct tagstruct*t) { + assert(t); + return t->rindex >= t->length; +} + +const uint8_t* tagstruct_data(struct tagstruct*t, size_t *l) { + assert(t && t->dynamic && l); + *l = t->length; + return t->data; +} + diff --git a/src/tagstruct.h b/src/tagstruct.h new file mode 100644 index 00000000..5572c64c --- /dev/null +++ b/src/tagstruct.h @@ -0,0 +1,30 @@ +#ifndef footagstructhfoo +#define footagstructhfoo + +#include +#include + +#include "sample.h" + +struct tagstruct; + +struct tagstruct *tagstruct_new(const uint8_t* data, size_t length); +void tagstruct_free(struct tagstruct*t); +uint8_t* tagstruct_free_data(struct tagstruct*t, size_t *l); + +void tagstruct_puts(struct tagstruct*t, const char *s); +void tagstruct_putu32(struct tagstruct*t, uint32_t i); +void tagstruct_putu8(struct tagstruct*t, uint8_t c); +void tagstruct_put_sample_spec(struct tagstruct *t, struct sample_spec *ss); + +int tagstruct_gets(struct tagstruct*t, const char **s); +int tagstruct_getu32(struct tagstruct*t, uint32_t *i); +int tagstruct_getu8(struct tagstruct*t, uint8_t *c); +int tagstruct_get_sample_spec(struct tagstruct *t, struct sample_spec *ss); + +int tagstruct_eof(struct tagstruct*t); +const uint8_t* tagstruct_data(struct tagstruct*t, size_t *l); + + + +#endif diff --git a/src/todo b/src/todo index 87b5c304..aeb7ae5f 100644 --- a/src/todo +++ b/src/todo @@ -1,4 +1,3 @@ -- cli protocol - native protocol/library - simple control protocol: kill client/input/output; set_volume - resampling -- cgit