updating tests

2 files changed, 312 insertions, 298 deletions

diff --git a/test/test_itc.c b/test/test_itc.c
index 779a10b..6af69c0 100644
--- a/test/test_itc.c
+++ b/test/test_itc.c
@@ -7,166 +7,166 @@
 
 static void test_itc_alloc_free(void)
 {
-        itc *ctx;
-        TEST((ctx = itc_alloc(0)) == NULL);
-        TEST((ctx = itc_alloc(1)) != NULL);
-        itc_free(&ctx, NULL);
-        TEST(ctx == NULL);
+	itc *ctx;
+	TEST((ctx = itc_alloc(0)) == NULL);
+	TEST((ctx = itc_alloc(1)) != NULL);
+	itc_free(&ctx, NULL);
+	TEST(ctx == NULL);
 }
 
 static void test_itc_inject_retrive(void)
 {
-        unsigned int i;
-        unsigned int *e[3] = {NULL};
-        int *r;
-        itc *ctx;
-
-        for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
-                e[i] = malloc(sizeof(*e[i]));
-                *e[i] = i;
-        }
-
-        ctx = itc_alloc(3);
-        TEST(itc_inject(ctx, 0, e[0]) == 0);
-        TEST(itc_inject(ctx, 0, e[1]) == 0);
-        TEST(itc_inject(ctx, 0, e[2]) == 0);
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 0);
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 1);
-        TEST(itc_inject(ctx, 0, e[0]) == 0);
-        TEST(itc_inject(ctx, 0, e[1]) == 0);
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 2);
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 0);
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 1);
-        TEST(itc_retrive(ctx, 0) == NULL);
-        TEST(itc_inject(ctx, 0, e[0]) == 0);
-        itc_free(&ctx, free);
-
-        e[0] = malloc(sizeof(*e[0]));
-        *e[0] = 0;
-        ctx = itc_alloc(2);
-        TEST(itc_inject(ctx, 0, e[0]) == 0);
-        TEST(itc_inject(ctx, 0, e[1]) == 0);
-        TEST(itc_inject(ctx, 0, e[2]));
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 0);
-        TEST((r = itc_retrive(ctx, 0)) != NULL);
-        TEST(*r == 1);
-        TEST(itc_retrive(ctx, 0) == NULL);
-        itc_free(&ctx, free);
-        for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
-                free(e[i]);
-        }
+	unsigned int i;
+	unsigned int *e[3] = {NULL};
+	int *r;
+	itc *ctx;
+
+	for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
+		e[i] = malloc(sizeof(*e[i]));
+		*e[i] = i;
+	}
+
+	ctx = itc_alloc(3);
+	TEST(itc_inject(ctx, 0, e[0]) == 0);
+	TEST(itc_inject(ctx, 0, e[1]) == 0);
+	TEST(itc_inject(ctx, 0, e[2]) == 0);
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 0);
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 1);
+	TEST(itc_inject(ctx, 0, e[0]) == 0);
+	TEST(itc_inject(ctx, 0, e[1]) == 0);
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 2);
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 0);
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 1);
+	TEST(itc_retrive(ctx, 0) == NULL);
+	TEST(itc_inject(ctx, 0, e[0]) == 0);
+	itc_free(&ctx, free);
+
+	e[0] = malloc(sizeof(*e[0]));
+	*e[0] = 0;
+	ctx = itc_alloc(2);
+	TEST(itc_inject(ctx, 0, e[0]) == 0);
+	TEST(itc_inject(ctx, 0, e[1]) == 0);
+	TEST(itc_inject(ctx, 0, e[2]));
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 0);
+	TEST((r = itc_retrive(ctx, 0)) != NULL);
+	TEST(*r == 1);
+	TEST(itc_retrive(ctx, 0) == NULL);
+	itc_free(&ctx, free);
+	for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
+		free(e[i]);
+	}
 }
 
 static void test_itc_queued_slots(void)
 {
-        itc *ctx;
-        ctx = itc_alloc(3);
-        TEST(itc_get_queued(NULL) < 0);
-        TEST(itc_get_queued(ctx) == 0);
-        TEST(itc_get_slots(NULL) < 0);
-        TEST(itc_get_slots(ctx) == 3);
-        itc_inject(ctx, 0, NULL);
-        TEST(itc_get_queued(ctx) == 0);
-        TEST(itc_get_slots(ctx) == 3);
-        itc_inject(ctx, 0, ctx);
-        TEST(itc_get_queued(ctx) == 1);
-        TEST(itc_get_slots(ctx) == 3);
-        itc_inject(ctx, 0, ctx);
-        TEST(itc_get_queued(ctx) == 2);
-        TEST(itc_get_slots(ctx) == 3);
-        itc_retrive(ctx, 0);
-        TEST(itc_get_queued(ctx) == 1);
-        TEST(itc_get_slots(ctx) == 3);
-        itc_retrive(ctx, 0);
-        TEST(itc_get_queued(ctx) == 0);
-        TEST(itc_get_slots(ctx) == 3);
-        itc_free(&ctx, NULL);
+	itc *ctx;
+	ctx = itc_alloc(3);
+	TEST(itc_get_queued(NULL) < 0);
+	TEST(itc_get_queued(ctx) == 0);
+	TEST(itc_get_slots(NULL) < 0);
+	TEST(itc_get_slots(ctx) == 3);
+	itc_inject(ctx, 0, NULL);
+	TEST(itc_get_queued(ctx) == 0);
+	TEST(itc_get_slots(ctx) == 3);
+	itc_inject(ctx, 0, ctx);
+	TEST(itc_get_queued(ctx) == 1);
+	TEST(itc_get_slots(ctx) == 3);
+	itc_inject(ctx, 0, ctx);
+	TEST(itc_get_queued(ctx) == 2);
+	TEST(itc_get_slots(ctx) == 3);
+	itc_retrive(ctx, 0);
+	TEST(itc_get_queued(ctx) == 1);
+	TEST(itc_get_slots(ctx) == 3);
+	itc_retrive(ctx, 0);
+	TEST(itc_get_queued(ctx) == 0);
+	TEST(itc_get_slots(ctx) == 3);
+	itc_free(&ctx, NULL);
 }
 
 static void *test_itc_wait_empty_th(void *_ctx)
 {
-        itc *ctx = _ctx;
-        int *e;
-
-        sleep(2);
-        TEST(ctx != NULL);
-        TEST((e = itc_retrive(ctx, 0)) != NULL);
-        TEST(*e == 0);
-        free(e);
-        TEST((e = itc_retrive(ctx, 0)) != NULL);
-        TEST(*e == 1);
-        free(e);
-        TEST((e = itc_retrive(ctx, 0)) != NULL);
-        TEST(*e == 2);
-        free(e);
-        TEST(itc_retrive(ctx, 0) == NULL);
-
-        return NULL;
+	itc *ctx = _ctx;
+	int *e;
+
+	sleep(2);
+	TEST(ctx != NULL);
+	TEST((e = itc_retrive(ctx, 0)) != NULL);
+	TEST(*e == 0);
+	free(e);
+	TEST((e = itc_retrive(ctx, 0)) != NULL);
+	TEST(*e == 1);
+	free(e);
+	TEST((e = itc_retrive(ctx, 0)) != NULL);
+	TEST(*e == 2);
+	free(e);
+	TEST(itc_retrive(ctx, 0) == NULL);
+
+	return NULL;
 }
 
 static void test_itc_wait_empty(void)
 {
-        itc *ctx;
-        pthread_t th;
-        int *e[3];
-        int i;
-
-        for (i = 0; i < 3; i++) {
-                e[i] = malloc(sizeof(int));
-                *e[i] = i;
-        }
-
-        ctx = itc_alloc(3);
-        itc_inject(ctx, 0, e[0]);
-        itc_inject(ctx, 0, e[1]);
-        itc_inject(ctx, 0, e[2]);
-        pthread_create(&th, NULL, test_itc_wait_empty_th, ctx);
-        pthread_detach(th);
-        itc_wait_empty(ctx);
-        sleep(2);
-        TEST(itc_get_queued(ctx) == 0);
-        TEST(itc_retrive(ctx, 0) == NULL);
-        itc_free(&ctx, NULL);
+	itc *ctx;
+	pthread_t th;
+	int *e[3];
+	int i;
+
+	for (i = 0; i < 3; i++) {
+		e[i] = malloc(sizeof(int));
+		*e[i] = i;
+	}
+
+	ctx = itc_alloc(3);
+	itc_inject(ctx, 0, e[0]);
+	itc_inject(ctx, 0, e[1]);
+	itc_inject(ctx, 0, e[2]);
+	pthread_create(&th, NULL, test_itc_wait_empty_th, ctx);
+	pthread_detach(th);
+	itc_wait_empty(ctx);
+	sleep(2);
+	TEST(itc_get_queued(ctx) == 0);
+	TEST(itc_retrive(ctx, 0) == NULL);
+	itc_free(&ctx, NULL);
 }
 
 static void test_itc_discard_all(void)
 {
-        itc *ctx;
-        unsigned int *e[4];
-        unsigned int i;
-
-        for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
-                e[i] = malloc(sizeof(*e[i]));
-                *e[i] = i;
-        }
-
-        ctx = itc_alloc(5);
-        itc_discard_all(ctx, free);
-        TEST(itc_get_queued(ctx) == 0);
-        itc_inject(ctx, 0, e[0]);
-        itc_inject(ctx, 0, e[1]);
-        itc_inject(ctx, 0, e[2]);
-        itc_inject(ctx, 0, e[3]);
-        itc_discard_all(ctx, free);
-        TEST(itc_get_queued(ctx) == 0);
-        itc_free(&ctx, NULL);
+	itc *ctx;
+	unsigned int *e[4];
+	unsigned int i;
+
+	for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
+		e[i] = malloc(sizeof(*e[i]));
+		*e[i] = i;
+	}
+
+	ctx = itc_alloc(5);
+	itc_discard_all(ctx, free);
+	TEST(itc_get_queued(ctx) == 0);
+	itc_inject(ctx, 0, e[0]);
+	itc_inject(ctx, 0, e[1]);
+	itc_inject(ctx, 0, e[2]);
+	itc_inject(ctx, 0, e[3]);
+	itc_discard_all(ctx, free);
+	TEST(itc_get_queued(ctx) == 0);
+	itc_free(&ctx, NULL);
 }
 
 int main(int argc, char **argv)
 {
-        (void)argc;
-        (void)argv;
-        test_itc_alloc_free();
-        test_itc_inject_retrive();
-        test_itc_queued_slots();
-        test_itc_wait_empty();
-        test_itc_discard_all();
-        return 0;
+	(void)argc;
+	(void)argv;
+	test_itc_alloc_free();
+	test_itc_inject_retrive();
+	test_itc_queued_slots();
+	test_itc_wait_empty();
+	test_itc_discard_all();
+	return 0;
 }
 
diff --git a/test/test_tuberia.c b/test/test_tuberia.c
index 6fe62b5..12920d1 100644
--- a/test/test_tuberia.c
+++ b/test/test_tuberia.c
@@ -7,186 +7,200 @@
 
 static void *process(void *element, void *opaque)
 {
-        unsigned int *e = element;
-        unsigned int *r;
-        int *s = opaque;
-        r = malloc(sizeof(*r));
-        *r = *e + (s != NULL ? *s : 0);
-        free(e);
-        return r;
+	unsigned int *e = element;
+	unsigned int *r;
+	int *s = opaque;
+	r = malloc(sizeof(*r));
+	*r = *e + (s != NULL ? *s : 0);
+	free(e);
+	return r;
 }
 
 static void test_tube_source_stage_alloc_free(void)
 {
-        tube_source *source;
-        tube_stage *stage;
-        TEST((source = tube_source_alloc(0, NULL, NULL, free)) == NULL);
-        TEST((source = tube_source_alloc(1, NULL, NULL, free)) != NULL);
-        tube_source_and_stages_free(&source);
-        TEST(source == NULL);
-        TEST((source = tube_source_alloc(3, NULL, NULL, free)) != NULL);
-        tube_source_and_stages_free(&source);
-        TEST(source == NULL);
-        TEST((stage = tube_stage_alloc(0, NULL, NULL, free)) == NULL);
-        TEST((stage = tube_stage_alloc(3, NULL, NULL, free)) == NULL);
-        TEST((stage = tube_stage_alloc(0, process, NULL, free)) == NULL);
-        TEST((stage = tube_stage_alloc(1, process, NULL, free)) != NULL);
-        tube_stage_free(&stage);
-        TEST(stage == NULL);
-        TEST((stage = tube_stage_alloc(3, process, NULL, free)) != NULL);
-        tube_stage_free(&stage);
-        TEST(stage == NULL);
-        TEST((source = tube_source_alloc(1, NULL, NULL, free)) != NULL);
-        TEST(tube_stage_append(source, tube_stage_alloc(1, process, NULL, free)) == 0);
-        TEST(tube_stage_append(source, tube_stage_alloc(1, process, NULL, free)) == 0);
-        TEST(tube_stage_append(source, tube_stage_alloc(1, process, NULL, free)) == 0);
-        tube_source_and_stages_free(&source);
-        TEST(source == NULL);
+	tube_source *source;
+	tube_stage *stage;
+	TEST((source = tube_source_alloc(0, NULL, NULL, free)) == NULL);
+	TEST((source = tube_source_alloc(1, NULL, NULL, free)) != NULL);
+	tube_source_and_stages_free(&source);
+	TEST(source == NULL);
+	TEST((source = tube_source_alloc(3, NULL, NULL, free)) != NULL);
+	tube_source_and_stages_free(&source);
+	TEST(source == NULL);
+	TEST((stage = tube_stage_alloc(0, NULL, NULL, free)) == NULL);
+	TEST((stage = tube_stage_alloc(3, NULL, NULL, free)) == NULL);
+	TEST((stage = tube_stage_alloc(0, process, NULL, free)) == NULL);
+	TEST((stage = tube_stage_alloc(1, process, NULL, free)) != NULL);
+	tube_stage_free(&stage);
+	TEST(stage == NULL);
+	TEST((stage = tube_stage_alloc(3, process, NULL, free)) != NULL);
+	tube_stage_free(&stage);
+	TEST(stage == NULL);
+	TEST((source = tube_source_alloc(1, NULL, NULL, free)) != NULL);
+	TEST(tube_stage_append(source, tube_stage_alloc(1, process, NULL, free)) == 0);
+	TEST(tube_stage_append(source, tube_stage_alloc(1, process, NULL, free)) == 0);
+	TEST(tube_stage_append(source, tube_stage_alloc(1, process, NULL, free)) == 0);
+	tube_source_and_stages_free(&source);
+	TEST(source == NULL);
 }
 
 static void test_tube_alloc_free(void)
 {
-        tube *ctx;
-        tube_source *source;
-        source = tube_source_alloc(1, NULL, NULL, free);
-        tube_stage_append(source, tube_stage_alloc(1, process, NULL, free));
-        tube_stage_append(source, tube_stage_alloc(1, process, NULL, free));
-        tube_stage_append(source, tube_stage_alloc(1, process, NULL, free));
-        TEST((ctx = tube_alloc(NULL, NULL, NULL)) == NULL);
-        TEST((ctx = tube_alloc(source, NULL, NULL)) != NULL);
-        tube_source_and_stages_free(&source);
-        tube_free(&ctx);
-        TEST(ctx == NULL);
+	tube *ctx;
+	tube_source *source;
+	source = tube_source_alloc(1, NULL, NULL, free);
+	tube_stage_append(source, tube_stage_alloc(1, process, NULL, free));
+	tube_stage_append(source, tube_stage_alloc(1, process, NULL, free));
+	tube_stage_append(source, tube_stage_alloc(1, process, NULL, free));
+	TEST((ctx = tube_alloc(NULL, NULL, NULL)) == NULL);
+	TEST((ctx = tube_alloc(source, NULL, NULL)) != NULL);
+	tube_source_and_stages_free(&source);
+	tube_free(&ctx);
+	TEST(ctx == NULL);
 }
 
 static void test_tube_inject_retrive(void)
 {
-        tube *ctx;
-        tube_source *source;
-        int sum[] = {7, 5};
-        unsigned int *e[10];
-        unsigned int r[10] = {12, 13, 14, 15, 16, 17, 18, 19, 20, 21};
-        unsigned int *o;
-        unsigned int i;
-
-        source = tube_source_alloc(3, NULL, NULL, free);
-        tube_stage_append(source, tube_stage_alloc(2, process, &sum[0], free));
-        tube_stage_append(source, tube_stage_alloc(4, process, &sum[1], free));
-        ctx = tube_alloc(source, NULL, NULL);
-        tube_source_and_stages_free(&source);
-
-        for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
-                e[i] = malloc(sizeof(*e[i]));
-                *e[i] = i;
-        }
-        for (i = 0; i < 3; i++) {
-                TEST(tube_inject(ctx, 1000, e[i]) == 0);
-        }
-        TEST(tube_inject(ctx, 1000, e[i]) < 0);
-        TEST(tube_start(ctx) == 0);
-        for (; i < sizeof(e)/sizeof(e[0]); i++) {
-                TEST(tube_inject(ctx, 1000, e[i]) == 0);
-        }
-        for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
-                TEST((o = tube_retrive(ctx, 1000)) != NULL);
-                TEST(*o == r[i]);
-                free(o);
-        }
-        tube_free(&ctx);
+	tube *ctx;
+	tube_source *source;
+	int sum[] = {7, 5};
+	unsigned int *e[10];
+	unsigned int *e2[7];
+	unsigned int r[10] = {12, 13, 14, 15, 16, 17, 18, 19, 20, 21};
+	unsigned int *o;
+	unsigned int i;
+
+	source = tube_source_alloc(3, NULL, NULL, free);
+	tube_stage_append(source, tube_stage_alloc(2, process, &sum[0], free));
+	tube_stage_append(source, tube_stage_alloc(4, process, &sum[1], free));
+	ctx = tube_alloc(source, NULL, NULL);
+	tube_source_and_stages_free(&source);
+
+	for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
+		e[i] = malloc(sizeof(*e[i]));
+		*e[i] = i;
+	}
+	for (i = 0; i < 3; i++) {
+		TEST(tube_inject(ctx, 1000, e[i]) == 0);
+	}
+	TEST(tube_inject(ctx, 1000, e[i]) < 0);
+	TEST(tube_start(ctx) == 0);
+	for (; i < sizeof(e)/sizeof(e[0]); i++) {
+		TEST(tube_inject(ctx, 1000, e[i]) == 0);
+	}
+	for (i = 0; i < sizeof(e)/sizeof(e[0]); i++) {
+		TEST((o = tube_retrive(ctx, 1000)) != NULL);
+		TEST(*o == r[i]);
+		free(o);
+	}
+	for (i = 0; i < sizeof(e2)/sizeof(e2[0]); i++) {
+		e2[i] = malloc(sizeof(*e2[i]));
+		*e2[i] = i + sum[0];
+	}
+	for (i = 0; i < sizeof(e2)/sizeof(e2[0]); i++) {
+		TEST(tube_inject_at(ctx, 1, 1000, e2[i]) == 0);
+	}
+	TEST(tube_inject_at(ctx, 1, 1000, e2[0]) < 0);
+	for (i = 0; i < sizeof(e2)/sizeof(e2[0]); i++) {
+		TEST((o = tube_retrive(ctx, 1000)) != NULL);
+		TEST(*o == r[i]);
+		free(o);
+	}
+	tube_free(&ctx);
 }
 
 static void test_tube_get_queued_get_slots(void)
 {
-        tube *ctx;
-        tube_source *source;
-        unsigned int *e;
-
-        source = tube_source_alloc(2, NULL, NULL, free);
-        tube_stage_append(source, tube_stage_alloc(3, process, NULL, free));
-        ctx = tube_alloc(source, NULL, NULL);
-        tube_source_and_stages_free(&source);
-        tube_start(ctx);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 0);
-        TEST(tube_get_queued(ctx, 1) == 0);
-
-        e = malloc(sizeof(*e));
-        *e = 1;
-        tube_inject(ctx, 0, e);
-        e = malloc(sizeof(*e));
-        *e = 2;
-        tube_inject(ctx, 0, e);
-        sleep(2);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 0);
-        TEST(tube_get_queued(ctx, 1) == 2);
-
-        e = malloc(sizeof(*e));
-        *e = 3;
-        tube_inject(ctx, 0, e);
-        e = malloc(sizeof(*e));
-        *e = 4;
-        tube_inject(ctx, 0, e);
-        e = malloc(sizeof(*e));
-        *e = 5;
-        tube_inject(ctx, 1000, e);
-        sleep(2);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 1);
-        TEST(tube_get_queued(ctx, 1) == 3);
-
-        free(tube_retrive(ctx, 1000));
-        sleep(2);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 0);
-        TEST(tube_get_queued(ctx, 1) == 3);
-
-        free(tube_retrive(ctx, 1000));
-        sleep(2);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 0);
-        TEST(tube_get_queued(ctx, 1) == 3);
-
-        free(tube_retrive(ctx, 1000));
-        sleep(2);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 0);
-        TEST(tube_get_queued(ctx, 1) == 2);
-
-        free(tube_retrive(ctx, 0));
-        free(tube_retrive(ctx, 0));
-        sleep(2);
-
-        TEST(tube_get_slots(ctx, 0) == 2);
-        TEST(tube_get_slots(ctx, 1) == 3);
-        TEST(tube_get_queued(ctx, 0) == 0);
-        TEST(tube_get_queued(ctx, 1) == 0);
-
-        tube_free(&ctx);
+	tube *ctx;
+	tube_source *source;
+	unsigned int *e;
+
+	source = tube_source_alloc(2, NULL, NULL, free);
+	tube_stage_append(source, tube_stage_alloc(3, process, NULL, free));
+	ctx = tube_alloc(source, NULL, NULL);
+	tube_source_and_stages_free(&source);
+	tube_start(ctx);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 0);
+	TEST(tube_get_queued(ctx, 1) == 0);
+
+	e = malloc(sizeof(*e));
+	*e = 1;
+	tube_inject(ctx, 0, e);
+	e = malloc(sizeof(*e));
+	*e = 2;
+	tube_inject(ctx, 0, e);
+	sleep(2);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 0);
+	TEST(tube_get_queued(ctx, 1) == 2);
+
+	e = malloc(sizeof(*e));
+	*e = 3;
+	tube_inject(ctx, 0, e);
+	e = malloc(sizeof(*e));
+	*e = 4;
+	tube_inject(ctx, 0, e);
+	e = malloc(sizeof(*e));
+	*e = 5;
+	tube_inject(ctx, 1000, e);
+	sleep(2);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 1);
+	TEST(tube_get_queued(ctx, 1) == 3);
+
+	free(tube_retrive(ctx, 1000));
+	sleep(2);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 0);
+	TEST(tube_get_queued(ctx, 1) == 3);
+
+	free(tube_retrive(ctx, 1000));
+	sleep(2);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 0);
+	TEST(tube_get_queued(ctx, 1) == 3);
+
+	free(tube_retrive(ctx, 1000));
+	sleep(2);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 0);
+	TEST(tube_get_queued(ctx, 1) == 2);
+
+	free(tube_retrive(ctx, 0));
+	free(tube_retrive(ctx, 0));
+	sleep(2);
+
+	TEST(tube_get_slots(ctx, 0) == 2);
+	TEST(tube_get_slots(ctx, 1) == 3);
+	TEST(tube_get_queued(ctx, 0) == 0);
+	TEST(tube_get_queued(ctx, 1) == 0);
+
+	tube_free(&ctx);
 }
 
 int main(int argc, char **argv)
 {
-        (void)argc;
-        (void)argv;
-        test_tube_source_stage_alloc_free();
-        test_tube_alloc_free();
-        test_tube_inject_retrive();
-        test_tube_get_queued_get_slots();
-        /*test_tube_wait_empty();
-        test_tube_discard_all();*/
-        return 0;
+	(void)argc;
+	(void)argv;
+	test_tube_source_stage_alloc_free();
+	test_tube_alloc_free();
+	test_tube_inject_retrive();
+	test_tube_get_queued_get_slots();
+	/*test_tube_wait_empty();
+	  test_tube_discard_all();*/
+	return 0;
 }