#include "../src/tuberia.h"
#include "test.h"

#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>

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;
}

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);
}

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);
}

static void test_tube_inject_retrive(void)
{
	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);
}

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;
}