summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/Makefile.am27
-rw-r--r--src/module-protocol-stub.c13
-rw-r--r--src/packet.c21
-rw-r--r--src/packet.h6
-rw-r--r--src/protocol-native-tcp.c19
-rw-r--r--src/protocol-native-unix.c27
-rw-r--r--src/protocol-native.c349
-rw-r--r--src/protocol-native.h5
-rw-r--r--src/protocol-simple.c1
-rw-r--r--src/pstream.c95
-rw-r--r--src/pstream.h6
-rw-r--r--src/sample.h2
-rw-r--r--src/sinkinput.h1
-rw-r--r--src/tagstruct.c187
-rw-r--r--src/tagstruct.h30
-rw-r--r--src/todo1
16 files changed, 680 insertions, 110 deletions
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 <stdint.h>
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 <stdio.h>
+#include <assert.h>
+#include <stdlib.h>
+
#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 <stdlib.h>
#include <assert.h>
+#include <netinet/in.h>
#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 <stdlib.h>
+#include <string.h>
+#include <netinet/in.h>
+#include <assert.h>
+
+#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 <inttypes.h>
+#include <sys/types.h>
+
+#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
generated by cgit (git 2.34.1) at 2025-11-04 06:29:27 +0000