add resampling

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

src/Makefile.am

@@ -55,7 +55,8 @@ polypaudio_SOURCES = idxset.c idxset.h \
 		namereg.c namereg.h \
 		sconv.c sconv.h \
 		resampler.c resampler.h \
-		endianmacros.h
+		endianmacros.h \
+		memchunk.c memchunk.h
 polypaudio_CFLAGS = $(AM_CFLAGS) $(LIBSAMPLERATE_CFLAGS)
 polypaudio_INCLUDES = $(INCLTDL)
 polypaudio_LDADD = $(LIBLTDL) $(LIBSAMPLERATE_LIBS)
@@ -196,7 +197,8 @@ libpolyp_la_SOURCES = polyp.c polyp.h \
 		socket-client.c socket-client.h \
 		packet.c packet.h \
 		queue.c queue.h \
-		dynarray.c dynarray.h
+		dynarray.c dynarray.h \
+		memchunk.c memchunk.h
 
 pacat_SOURCES = pacat.c
 pacat_LDADD = libpolyp.la

src/iochannel.c

@@ -15,7 +15,7 @@ struct iochannel {
     
     int readable;
     int writable;
-
+    
     int no_close;
 
     void* input_source, *output_source;
@@ -147,8 +147,8 @@ ssize_t iochannel_write(struct iochannel*io, const void*data, size_t l) {
 ssize_t iochannel_read(struct iochannel*io, void*data, size_t l) {
     ssize_t r;
     
-    assert(io && data && l && io->ifd >= 0);
-
+    assert(io && data && io->ifd >= 0);
+    
     if ((r = read(io->ifd, data, l)) >= 0) {
         io->readable = 0;
         enable_mainloop_sources(io);

src/main.c

@@ -38,15 +38,15 @@ int main(int argc, char *argv[]) {
     c = core_new(pa_mainloop_get_api(mainloop));
     assert(c);
 
-    module_load(c, "module-oss-mmap", "/dev/dsp1");
-/*    module_load(c, "module-pipe-sink", NULL);
+    module_load(c, "module-oss-mmap", "/dev/dsp");
+/*    module_load(c, "module-pipe-sink", NULL);*/
     module_load(c, "module-simple-protocol-tcp", NULL);
-    module_load(c, "module-simple-protocol-unix", NULL);
+/*    module_load(c, "module-simple-protocol-unix", NULL);
     module_load(c, "module-cli-protocol-tcp", NULL);
     module_load(c, "module-cli-protocol-unix", NULL);
-    module_load(c, "module-native-protocol-tcp", NULL);
-    module_load(c, "module-native-protocol-unix", NULL);*/
-    module_load(c, "module-esound-protocol-tcp", NULL);
+    module_load(c, "module-native-protocol-tcp", NULL);*/
+    module_load(c, "module-native-protocol-unix", NULL);
+/*    module_load(c, "module-esound-protocol-tcp", NULL);*/
     module_load(c, "module-cli", NULL);
     
     fprintf(stderr, "main: mainloop entry.\n");

src/memblock.c

@@ -78,17 +78,3 @@ void memblock_unref_fixed(struct memblock *b) {
     b->type = MEMBLOCK_DYNAMIC;
 }
 
-void memchunk_make_writable(struct memchunk *c) {
-    struct memblock *n;
-    assert(c && c->memblock && c->memblock->ref >= 1);
-
-    if (c->memblock->ref == 1)
-        return;
-    
-    n = memblock_new(c->length);
-    assert(n);
-    memcpy(n->data, c->memblock->data+c->index, c->length);
-    memblock_unref(c->memblock);
-    c->memblock = n;
-    c->index = 0;
-}

src/memblock.h

@@ -13,11 +13,6 @@ struct memblock {
     void *data;
 };
 
-struct memchunk {
-    struct memblock *memblock;
-    size_t index, length;
-};
-
 struct memblock *memblock_new(size_t length);
 struct memblock *memblock_new_fixed(void *data, size_t length);
 struct memblock *memblock_new_dynamic(void *data, size_t length);
@@ -29,8 +24,6 @@ void memblock_unref_fixed(struct memblock*b);
 
 #define memblock_assert_exclusive(b) assert((b)->ref == 1)
 
-void memchunk_make_writable(struct memchunk *c);
-
 extern unsigned memblock_count, memblock_total;
 
 #endif

src/memblockq.c

@@ -18,6 +18,7 @@ struct memblockq {
     size_t total_length, maxlength, base, prebuf;
     int measure_delay;
     uint32_t delay;
+    struct mcalign *mcalign;
 };
 
 struct memblockq* memblockq_new(size_t maxlength, size_t base, size_t prebuf) {
@@ -40,6 +41,8 @@ struct memblockq* memblockq_new(size_t maxlength, size_t base, size_t prebuf) {
 
     bq->measure_delay = 0;
     bq->delay = 0;
+
+    bq->mcalign = NULL;
     
     return bq;
 }
@@ -48,6 +51,9 @@ void memblockq_free(struct memblockq* bq) {
     struct memblock_list *l;
     assert(bq);
 
+    if (bq->mcalign)
+        mcalign_free(bq->mcalign);
+
     while ((l = bq->blocks)) {
         bq->blocks = l->next;
         memblock_unref(l->chunk.memblock);
@@ -57,9 +63,9 @@ void memblockq_free(struct memblockq* bq) {
     free(bq);
 }
 
-void memblockq_push(struct memblockq* bq, struct memchunk *chunk, size_t delta) {
+void memblockq_push(struct memblockq* bq, const struct memchunk *chunk, size_t delta) {
     struct memblock_list *q;
-    assert(bq && chunk && chunk->memblock && chunk->length);
+    assert(bq && chunk && chunk->memblock && chunk->length && (chunk->length % bq->base) == 0);
 
     q = malloc(sizeof(struct memblock_list));
     assert(q);
@@ -97,9 +103,14 @@ int memblockq_peek(struct memblockq* bq, struct memchunk *chunk) {
 
     *chunk = bq->blocks->chunk;
     memblock_ref(chunk->memblock);
+
+    if (chunk->memblock->ref != 2)
+        fprintf(stderr, "block %p with ref %u peeked.\n", chunk->memblock, chunk->memblock->ref);
+    
     return 0;
 }
 
+/*
 int memblockq_pop(struct memblockq* bq, struct memchunk *chunk) {
     struct memblock_list *q;
     
@@ -121,6 +132,7 @@ int memblockq_pop(struct memblockq* bq, struct memchunk *chunk) {
     free(q);
     return 0;
 }
+*/
 
 static uint32_t age(struct timeval *tv) {
     assert(tv);
@@ -143,7 +155,7 @@ static uint32_t age(struct timeval *tv) {
 }
 
 void memblockq_drop(struct memblockq *bq, size_t length) {
-    assert(bq);
+    assert(bq && length && (length % bq->base) == 0);
 
     while (length > 0) {
         size_t l = length;
@@ -229,3 +241,26 @@ uint32_t memblockq_missing_to(struct memblockq *bq, size_t qlen) {
 
     return qlen - bq->total_length;
 }
+
+void memblockq_push_align(struct memblockq* bq, const struct memchunk *chunk, size_t delta) {
+    struct memchunk rchunk;
+    assert(bq && chunk && bq->base);
+
+    if (bq->base == 1) {
+        memblockq_push(bq, chunk, delta);
+        return;
+    }
+
+    if (!bq->mcalign) {
+        bq->mcalign = mcalign_new(bq->base);
+        assert(bq->mcalign);
+    }
+    
+    mcalign_push(bq->mcalign, chunk);
+
+    while (mcalign_pop(bq->mcalign, &rchunk) >= 0) {
+        memblockq_push(bq, &rchunk, delta);
+        memblock_unref(rchunk.memblock);
+        delta = 0;
+    }
+}

src/memblockq.h

@@ -4,27 +4,48 @@
 #include <sys/types.h>
 
 #include "memblock.h"
+#include "memchunk.h"
 
 struct memblockq;
 
+/* Parameters: the maximum length of the memblock queue, a base value
+for all operations (that is, all byte operations shall work on
+multiples of this base value) and an amount of bytes to prebuffer
+before having memblockq_peek() succeed. */
 struct memblockq* memblockq_new(size_t maxlength, size_t base, size_t prebuf);
-void memblockq_free(struct memblockq* bq);
+void memblockq_free(struct memblockq*bq);
 
-void memblockq_push(struct memblockq* bq, struct memchunk *chunk, size_t delta);
+/* Push a new memory chunk into the queue. Optionally specify a value for future cancellation. This is currently not implemented, however! */
+void memblockq_push(struct memblockq* bq, const struct memchunk *chunk, size_t delta);
 
-int memblockq_pop(struct memblockq* bq, struct memchunk *chunk);
+/* Same as memblockq_push(), however chunks are filtered through a mcalign object, and thus aligned to multiples of base */
+void memblockq_push_align(struct memblockq* bq, const struct memchunk *chunk, size_t delta);
+
+/* Return a copy of the next memory chunk in the queue. It is not removed from the queue */
 int memblockq_peek(struct memblockq* bq, struct memchunk *chunk);
+
+/* Drop the specified bytes from the queue */
 void memblockq_drop(struct memblockq *bq, size_t length);
 
+/* Shorten the memblockq to the specified length by dropping data at the end of the queue */
 void memblockq_shorten(struct memblockq *bq, size_t length);
+
+/* Empty the memblockq */
 void memblockq_empty(struct memblockq *bq);
 
+/* Test if the memblockq is currently readable, that is, more data than base */
 int memblockq_is_readable(struct memblockq *bq);
+
+/* Test if the memblockq is currently writable for the specified amount of bytes */
 int memblockq_is_writable(struct memblockq *bq, size_t length);
 
+/* The time memory chunks stay in the queue until they are removed completely in usecs */
 uint32_t memblockq_get_delay(struct memblockq *bq);
+
+/* Return the length of the queue in bytes */
 uint32_t memblockq_get_length(struct memblockq *bq);
 
+/* Return how many bytes are missing in queue to the specified fill amount */
 uint32_t memblockq_missing_to(struct memblockq *bq, size_t qlen);
 
 #endif

src/pacat.c

@@ -77,7 +77,7 @@ static void context_complete_callback(struct pa_context *c, int success, void *u
     static const struct pa_sample_spec ss = {
         .format = PA_SAMPLE_S16NE,
         .rate = 44100,
-        .channels = 2
+        .channels = 1
     };
         
     assert(c && !stream);

src/polyp.c

@@ -15,7 +15,7 @@
 #define DEFAULT_MAX_LENGTH 20480
 #define DEFAULT_PREBUF 4096
 #define DEFAULT_TIMEOUT (5*60)
-#define DEFAULT_SERVER "/tmp/polypaudio_native"
+#define DEFAULT_SERVER "/tmp/polypaudio/native"
 
 struct pa_context {
     char *name;

src/protocol-esound.c

@@ -210,7 +210,7 @@ static int esd_proto_stream_play(struct connection *c, const void *data, size_t
     assert(c->input_memblockq);
 
     assert(!c->sink_input);
-    c->sink_input = sink_input_new(sink, &ss, name);
+    c->sink_input = sink_input_new(sink, name, &ss);
     assert(c->sink_input);
 
     c->sink_input->peek = sink_input_peek_cb;
@@ -218,7 +218,7 @@ static int esd_proto_stream_play(struct connection *c, const void *data, size_t
     c->sink_input->kill = sink_input_kill_cb;
     c->sink_input->get_latency = sink_input_get_latency_cb;
     c->sink_input->userdata = c;
-    
+
     c->state = ESD_STREAMING_DATA;
 
     c->protocol->n_player++;
@@ -460,7 +460,7 @@ static int do_read(struct connection *c) {
         chunk.index = 0;
 
         assert(c->input_memblockq);
-        memblockq_push(c->input_memblockq, &chunk, 0);
+        memblockq_push_align(c->input_memblockq, &chunk, 0);
         memblock_unref(chunk.memblock);
         assert(c->sink_input);
         sink_notify(c->sink_input->sink);

src/protocol-native.c

@@ -89,7 +89,7 @@ static struct playback_stream* playback_stream_new(struct connection *c, struct
     s->connection = c;
     s->qlength = qlen;
     
-    s->sink_input = sink_input_new(sink, ss, name);
+    s->sink_input = sink_input_new(sink, name, ss);
     assert(s->sink_input);
     s->sink_input->peek = sink_input_peek_cb;
     s->sink_input->drop = sink_input_drop_cb;
@@ -323,7 +323,7 @@ static int memblock_callback(struct pstream *p, uint32_t channel, int32_t delta,
     else
         stream->requested_bytes -= chunk->length;
     
-    memblockq_push(stream->memblockq, chunk, delta);
+    memblockq_push_align(stream->memblockq, chunk, delta);
     assert(stream->sink_input);
     sink_notify(stream->sink_input->sink);
 

src/protocol-simple.c

@@ -25,14 +25,16 @@ struct protocol_simple {
     struct socket_server*server;
     struct idxset *connections;
     enum protocol_simple_mode mode;
+    struct pa_sample_spec sample_spec;
 };
 
 #define BUFSIZE PIPE_BUF
 
-static void free_connection(void *data, void *userdata) {
-    struct connection *c = data;
-    assert(data);
-    
+static void connection_free(struct connection *c) {
+    assert(c);
+
+    idxset_remove_by_data(c->protocol->connections, c, NULL);
+
     if (c->sink_input)
         sink_input_free(c->sink_input);
     if (c->source_output)
@@ -47,13 +49,6 @@ static void free_connection(void *data, void *userdata) {
         memblockq_free(c->output_memblockq);
     free(c);
 }
-
-static void destroy_connection(struct connection *c) {
-    assert(c && c->protocol);
-    idxset_remove_by_data(c->protocol->connections, c, NULL);
-    free_connection(c, NULL);
-}
-
 static int do_read(struct connection *c) {
     struct memchunk chunk;
     ssize_t r;
@@ -77,7 +72,7 @@ static int do_read(struct connection *c) {
     chunk.index = 0;
 
     assert(c->input_memblockq);
-    memblockq_push(c->input_memblockq, &chunk, 0);
+    memblockq_push_align(c->input_memblockq, &chunk, 0);
     memblock_unref(chunk.memblock);
     assert(c->sink_input);
     sink_notify(c->sink_input->sink);
@@ -132,12 +127,12 @@ static void sink_input_drop_cb(struct sink_input *i, size_t length) {
     memblockq_drop(c->input_memblockq, length);
     
     if (do_read(c) < 0)
-        destroy_connection(c);
+        connection_free(c);
 }
 
 static void sink_input_kill_cb(struct sink_input *i) {
     assert(i && i->userdata);
-    destroy_connection((struct connection *) i->userdata);
+    connection_free((struct connection *) i->userdata);
 }
 
 
@@ -149,26 +144,26 @@ static uint32_t sink_input_get_latency_cb(struct sink_input *i) {
 
 /*** source_output callbacks ***/
 
-static void source_output_push_cb(struct source_output *o, struct memchunk *chunk) {
+static void source_output_push_cb(struct source_output *o, const struct memchunk *chunk) {
     struct connection *c = o->userdata;
     assert(o && c && chunk);
 
     memblockq_push(c->output_memblockq, chunk, 0);
 
     if (do_write(c) < 0)
-        destroy_connection(c);
+        connection_free(c);
 }
 
 static void source_output_kill_cb(struct source_output *o) {
     assert(o && o->userdata);
-    destroy_connection((struct connection *) o->userdata);
+    connection_free((struct connection *) o->userdata);
 }
 
 /*** client callbacks ***/
 
 static void client_kill_cb(struct client *c) {
     assert(c && c->userdata);
-    destroy_connection((struct connection *) c->userdata);
+    connection_free((struct connection *) c->userdata);
 }
 
 /*** iochannel callbacks ***/
@@ -178,7 +173,7 @@ static void io_callback(struct iochannel*io, void *userdata) {
     assert(io && c && c->io == io);
 
     if (do_read(c) < 0 || do_write(c) < 0)
-        destroy_connection(c);
+        connection_free(c);
 }
 
 /*** socket_server callbacks */
@@ -212,14 +207,14 @@ static void on_connection(struct socket_server*s, struct iochannel *io, void *us
             goto fail;
         }
 
-        c->source_output = source_output_new(source, &DEFAULT_SAMPLE_SPEC, c->client->name);
+        c->source_output = source_output_new(source, c->client->name, &p->sample_spec);
         assert(c->source_output);
         c->source_output->push = source_output_push_cb;
         c->source_output->kill = source_output_kill_cb;
         c->source_output->userdata = c;
 
         l = 5*pa_bytes_per_second(&DEFAULT_SAMPLE_SPEC); /* 5s */
-        c->output_memblockq = memblockq_new(l, pa_sample_size(&DEFAULT_SAMPLE_SPEC), l/2);
+        c->output_memblockq = memblockq_new(l, pa_sample_size(&p->sample_spec), l/2);
     }
 
     if (p->mode & PROTOCOL_SIMPLE_PLAYBACK) {
@@ -231,7 +226,7 @@ static void on_connection(struct socket_server*s, struct iochannel *io, void *us
             goto fail;
         }
 
-        c->sink_input = sink_input_new(sink, &DEFAULT_SAMPLE_SPEC, c->client->name);
+        c->sink_input = sink_input_new(sink, c->client->name, &p->sample_spec);
         assert(c->sink_input);
         c->sink_input->peek = sink_input_peek_cb;
         c->sink_input->drop = sink_input_drop_cb;
@@ -240,7 +235,7 @@ static void on_connection(struct socket_server*s, struct iochannel *io, void *us
         c->sink_input->userdata = c;
 
         l = pa_bytes_per_second(&DEFAULT_SAMPLE_SPEC)/2; /* half a second */
-        c->input_memblockq = memblockq_new(l, pa_sample_size(&DEFAULT_SAMPLE_SPEC), l/2);
+        c->input_memblockq = memblockq_new(l, pa_sample_size(&p->sample_spec), l/2);
     }
 
 
@@ -249,11 +244,8 @@ static void on_connection(struct socket_server*s, struct iochannel *io, void *us
     return;
     
 fail:
-    if (c) {
-        free_connection(c, NULL);
-        iochannel_free(c->io);
-        free(c);
-    }
+    if (c)
+        connection_free(c);
 }
 
 struct protocol_simple* protocol_simple_new(struct core *core, struct socket_server *server, enum protocol_simple_mode mode) {
@@ -266,6 +258,7 @@ struct protocol_simple* protocol_simple_new(struct core *core, struct socket_ser
     p->server = server;
     p->connections = idxset_new(NULL, NULL);
     p->mode = mode;
+    p->sample_spec = DEFAULT_SAMPLE_SPEC;
 
     socket_server_set_callback(p->server, on_connection, p);
     
@@ -274,9 +267,15 @@ struct protocol_simple* protocol_simple_new(struct core *core, struct socket_ser
 
 
 void protocol_simple_free(struct protocol_simple *p) {
+    struct connection *c;
     assert(p);
 
-    idxset_free(p->connections, free_connection, NULL);
+    while((c = idxset_first(p->connections, NULL)))
+        connection_free(c);
+
+    idxset_free(p->connections, NULL, NULL);
+    
     socket_server_free(p->server);
     free(p);
 }
+

src/pstream.h

@@ -7,6 +7,7 @@
 #include "memblock.h"
 #include "iochannel.h"
 #include "mainloop-api.h"
+#include "memchunk.h"
 
 struct pstream;
 

src/resampler.c

@@ -30,12 +30,13 @@ struct resampler* resampler_new(const struct pa_sample_spec *a, const struct pa_
     if (a->format == PA_SAMPLE_ALAW || a->format == PA_SAMPLE_ULAW || b->format == PA_SAMPLE_ALAW || b->format == PA_SAMPLE_ULAW)
         goto fail;
 
+    r = malloc(sizeof(struct resampler));
+    assert(r);
+
     r->channels = a->channels;
     if (b->channels < r->channels)
         r->channels = b->channels;
     
-    r = malloc(sizeof(struct resampler));
-    assert(r);
     r->i_buf = r->o_buf = NULL;
     r->i_alloc = r->o_alloc = 0;
 
@@ -82,11 +83,10 @@ size_t resampler_request(struct resampler *r, size_t out_length) {
 }
 
 
-int resampler_run(struct resampler *r, struct memchunk *in, struct memchunk *out) {
+void resampler_run(struct resampler *r, const struct memchunk *in, struct memchunk *out) {
     unsigned i_nchannels, o_nchannels, ins, ons, eff_ins, eff_ons;
     float *cbuf;
-    size_t in_bytes_used = 0;
-    assert(r && in && out && in->length && in->memblock);
+    assert(r && in && out && in->length && in->memblock && (in->length % r->i_sz) == 0);
 
     /* How many input samples? */
     ins = in->length/r->i_sz;
@@ -138,8 +138,8 @@ int resampler_run(struct resampler *r, struct memchunk *in, struct memchunk *out
         
         ret = src_process(r->src_state, &data);
         assert(ret == 0);
-
-        in_bytes_used = data.input_frames_used*r->i_sz;
+        assert((unsigned) data.input_frames_used == ins);
+        
         cbuf = r->o_buf;
         ons = data.output_frames_gen;
 
@@ -147,16 +147,9 @@ int resampler_run(struct resampler *r, struct memchunk *in, struct memchunk *out
             eff_ons = ons*r->o_ss.channels;
         else
             eff_ons = ons;
-    } else {
-        in_bytes_used = ins*r->i_sz;
+    } else
         cbuf = r->i_buf;
-    }
 
-    assert(in_bytes_used < in->length);
-    in->index += in_bytes_used;
-    in->length -= in_bytes_used;
-    
     r->from_float32_func(eff_ons, cbuf, out->memblock->data+out->index, o_nchannels);
     out->length = ons*r->o_sz;
-    return 0;
 }

src/resampler.h

@@ -3,6 +3,7 @@
 
 #include "sample.h"
 #include "memblock.h"
+#include "memchunk.h"
 
 struct resampler;
 
@@ -10,6 +11,6 @@ struct resampler* resampler_new(const struct pa_sample_spec *a, const struct pa_
 void resampler_free(struct resampler *r);
 
 size_t resampler_request(struct resampler *r, size_t out_length);
-int resampler_run(struct resampler *r, struct memchunk *in, struct memchunk *out);
+void resampler_run(struct resampler *r, const struct memchunk *in, struct memchunk *out);
 
 #endif

src/sample-util.h

@@ -3,6 +3,7 @@
 
 #include "sample.h"
 #include "memblock.h"
+#include "memchunk.h"
 
 #define DEFAULT_SAMPLE_SPEC default_sample_spec
 

src/sample.c

@@ -49,3 +49,9 @@ int pa_sample_spec_valid(const struct pa_sample_spec *spec) {
 
     return 1;
 }
+
+int pa_sample_spec_equal(const struct pa_sample_spec*a, const struct pa_sample_spec*b) {
+    assert(a && b);
+
+    return (a->format == b->format) && (a->rate == b->rate) && (a->channels == b->channels);
+}

src/sample.h

@@ -29,7 +29,7 @@ struct pa_sample_spec {
 size_t pa_bytes_per_second(const struct pa_sample_spec *spec);
 size_t pa_sample_size(const struct pa_sample_spec *spec);
 uint32_t pa_samples_usec(size_t length, const struct pa_sample_spec *spec);
-
 int pa_sample_spec_valid(const struct pa_sample_spec *spec);
+int pa_sample_spec_equal(const struct pa_sample_spec*a, const struct pa_sample_spec*b);
 
 #endif

src/sink.c

@@ -90,8 +90,7 @@ static unsigned fill_mix_info(struct sink *s, struct mix_info *info, unsigned ma
     assert(s && info);
 
     for (i = idxset_first(s->inputs, &index); maxinfo > 0 && i; i = idxset_next(s->inputs, &index)) {
-        assert(i->peek);
-        if (i->peek(i, &info->chunk) < 0)
+        if (sink_input_peek(i, &info->chunk) < 0)
             continue;
 
         info->volume = i->volume;
@@ -115,11 +114,10 @@ static void inputs_drop(struct sink *s, struct mix_info *info, unsigned maxinfo,
         assert(i && info->chunk.memblock);
         
         memblock_unref(info->chunk.memblock);
-        assert(i->drop);
-        i->drop(i, length);
+        sink_input_drop(i, length);
     }
 }
-
+        
 int sink_render(struct sink*s, size_t length, struct memchunk *result) {
     struct mix_info info[MAX_MIX_CHANNELS];
     unsigned n;

src/sinkinput.c

@@ -6,11 +6,18 @@
 #include "strbuf.h"
 #include "sample-util.h"
 
-struct sink_input* sink_input_new(struct sink *s, struct pa_sample_spec *spec, const char *name) {
+#define CONVERT_BUFFER_LENGTH 4096
+
+struct sink_input* sink_input_new(struct sink *s, const char *name, const struct pa_sample_spec *spec) {
     struct sink_input *i;
+    struct resampler *resampler = NULL;
     int r;
     assert(s && spec);
 
+    if (!pa_sample_spec_equal(spec, &s->sample_spec))
+        if (!(resampler = resampler_new(spec, &s->sample_spec)))
+            return NULL;
+    
     i = malloc(sizeof(struct sink_input));
     assert(i);
     i->name = name ? strdup(name) : NULL;
@@ -25,12 +32,16 @@ struct sink_input* sink_input_new(struct sink *s, struct pa_sample_spec *spec, c
 
     i->volume = VOLUME_NORM;
 
+    i->resampled_chunk.memblock = NULL;
+    i->resampled_chunk.index = i->resampled_chunk.length = 0;
+    i->resampler = resampler;
+    
     assert(s->core);
     r = idxset_put(s->core->sink_inputs, i, &i->index);
     assert(r == 0 && i->index != IDXSET_INVALID);
     r = idxset_put(s->inputs, i, NULL);
     assert(r == 0);
-    
+
     return i;    
 }
 
@@ -40,6 +51,11 @@ void sink_input_free(struct sink_input* i) {
     assert(i->sink && i->sink->core);
     idxset_remove_by_data(i->sink->core->sink_inputs, i, NULL);
     idxset_remove_by_data(i->sink->inputs, i, NULL);
+
+    if (i->resampled_chunk.memblock)
+        memblock_unref(i->resampled_chunk.memblock);
+    if (i->resampler)
+        resampler_free(i->resampler);
     
     free(i->name);
     free(i);
@@ -89,3 +105,53 @@ uint32_t sink_input_get_latency(struct sink_input *i) {
     return l;
 }
 
+
+int sink_input_peek(struct sink_input *i, struct memchunk *chunk) {
+    assert(i && chunk && i->peek && i->drop);
+
+    if (!i->resampler)
+        return i->peek(i, chunk);
+
+    if (!i->resampled_chunk.memblock) {
+        struct memchunk tchunk;
+        size_t l;
+        int ret;
+        
+        if ((ret = i->peek(i, &tchunk)) < 0)
+            return ret;
+
+        l = resampler_request(i->resampler, CONVERT_BUFFER_LENGTH);
+        if (tchunk.length > l)
+            tchunk.length = l;
+
+        i->drop(i, tchunk.length);
+        
+        resampler_run(i->resampler, &tchunk, &i->resampled_chunk);
+        memblock_unref(tchunk.memblock);
+    }
+
+    assert(i->resampled_chunk.memblock && i->resampled_chunk.length);
+    *chunk = i->resampled_chunk;
+    memblock_ref(i->resampled_chunk.memblock);
+    return 0;
+}
+
+void sink_input_drop(struct sink_input *i, size_t length) {
+    assert(i && length);
+
+    if (!i->resampler) {
+        i->drop(i, length);
+        return;
+    }
+    
+    assert(i->resampled_chunk.memblock && i->resampled_chunk.length >= length);
+
+    i->resampled_chunk.index += length;
+    i->resampled_chunk.length -= length;
+
+    if (!i->resampled_chunk.length) {
+        memblock_unref(i->resampled_chunk.memblock);
+        i->resampled_chunk.memblock = NULL;
+        i->resampled_chunk.index = i->resampled_chunk.length = 0;
+    }
+}

src/sinkinput.h

@@ -6,6 +6,7 @@
 #include "sink.h"
 #include "sample.h"
 #include "memblockq.h"
+#include "resampler.h"
 
 struct sink_input {
     uint32_t index;
@@ -21,9 +22,12 @@ struct sink_input {
     uint32_t (*get_latency) (struct sink_input *i);
 
     void *userdata;
+
+    struct memchunk resampled_chunk;
+    struct resampler *resampler;
 };
 
-struct sink_input* sink_input_new(struct sink *s, struct pa_sample_spec *spec, const char *name);
+struct sink_input* sink_input_new(struct sink *s, const char *name, const struct pa_sample_spec *spec);
 void sink_input_free(struct sink_input* i);
 
 /* Code that didn't create the input stream should call this function to
@@ -33,7 +37,7 @@ void sink_input_kill(struct sink_input *i);
 uint32_t sink_input_get_latency(struct sink_input *i);
 char *sink_input_list_to_string(struct core *c);
 
-
-
+int sink_input_peek(struct sink_input *i, struct memchunk *chunk);
+void sink_input_drop(struct sink_input *i, size_t length);
 
 #endif

src/source.c

@@ -70,7 +70,7 @@ static int do_post(void *p, uint32_t index, int *del, void*userdata) {
     struct source_output *o = p;
     assert(o && o->push && del && chunk);
 
-    o->push(o, chunk);
+    source_output_push(o, chunk);
     return 0;
 }
 

src/source.h

@@ -8,6 +8,7 @@ struct source;
 #include "sample.h"
 #include "idxset.h"
 #include "memblock.h"
+#include "memchunk.h"
 
 struct source {
     uint32_t index;

src/sourceoutput.c

@@ -5,11 +5,16 @@
 #include "sourceoutput.h"
 #include "strbuf.h"
 
-struct source_output* source_output_new(struct source *s, struct pa_sample_spec *spec, const char *name) {
+struct source_output* source_output_new(struct source *s, const char *name, const struct pa_sample_spec *spec) {
     struct source_output *o;
+    struct resampler *resampler = NULL;
     int r;
     assert(s && spec);
 
+    if (!pa_sample_spec_equal(&s->sample_spec, spec))
+        if (!(resampler = resampler_new(&s->sample_spec, spec)))
+            return NULL;
+    
     o = malloc(sizeof(struct source_output));
     assert(o);
     o->name = name ? strdup(name) : NULL;
@@ -19,6 +24,7 @@ struct source_output* source_output_new(struct source *s, struct pa_sample_spec
     o->push = NULL;
     o->kill = NULL;
     o->userdata = NULL;
+    o->resampler = resampler;
     
     assert(s->core);
     r = idxset_put(s->core->source_outputs, o, &o->index);
@@ -35,6 +41,9 @@ void source_output_free(struct source_output* o) {
     assert(o->source && o->source->core);
     idxset_remove_by_data(o->source->core->source_outputs, o, NULL);
     idxset_remove_by_data(o->source->outputs, o, NULL);
+
+    if (o->resampler)
+        resampler_free(o->resampler);
     
     free(o->name);
     free(o);
@@ -68,3 +77,21 @@ char *source_output_list_to_string(struct core *c) {
     
     return strbuf_tostring_free(s);
 }
+
+void source_output_push(struct source_output *o, const struct memchunk *chunk) {
+    struct memchunk rchunk;
+    assert(o && chunk && chunk->length && o->push);
+
+    if (!o->resampler) {
+        o->push(o, chunk);
+        return;
+    }
+
+    resampler_run(o->resampler, chunk, &rchunk);
+    if (!rchunk.length)
+        return;
+    
+    assert(rchunk.memblock);
+    o->push(o, &rchunk);
+    memblock_unref(rchunk.memblock);
+}

src/sourceoutput.h

@@ -6,6 +6,7 @@
 #include "source.h"
 #include "sample.h"
 #include "memblockq.h"
+#include "resampler.h"
 
 struct source_output {
     uint32_t index;
@@ -14,17 +15,21 @@ struct source_output {
     struct source *source;
     struct pa_sample_spec sample_spec;
     
-    void (*push)(struct source_output *o, struct memchunk *chunk);
+    void (*push)(struct source_output *o, const struct memchunk *chunk);
     void (*kill)(struct source_output* o);
 
+    struct resampler* resampler;
+    
     void *userdata;
 };
 
-struct source_output* source_output_new(struct source *s, struct pa_sample_spec *spec, const char *name);
+struct source_output* source_output_new(struct source *s, const char *name, const struct pa_sample_spec *spec);
 void source_output_free(struct source_output* o);
 
 void source_output_kill(struct source_output*o);
 
 char *source_output_list_to_string(struct core *c);
 
+void source_output_push(struct source_output *o, const struct memchunk *chunk);
+
 #endif

src/todo

@@ -1,20 +1,17 @@
+- recording (general, simple, esound, native)
+- make it embedable (add pa_ prefix too all identifiers)
 - native library/protocol:
-       recording
        sync() function
        more functions
-- esound protocol:
-       recording
-- move more stuff from module-oss[-dma] to liboss-util
 - simple library
-- simple control protocol: 
-       kill client/input/output
 - kill() routines in all modules
-- resampling
+- command line protocol: 
+       kill client/input/output
 - config parser/cmdline
-- record testing
-- mixing/volume
+- move more stuff from module-oss[-dma] to liboss-util
+- svn-id and license in every file
 
--- 0.1
+-- post 0.1
 - future cancellation
 - client-ui
 - clip cache