partial implementation of native protocol

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@30 fefdeb5f-60dc-0310-8127-8f9354f1896f

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