bshell/bshell.runtime/runtime/pipeline.c

#include <bshell/runtime/cmdcall.h>
#include <bshell/runtime/pipeline.h>
#include <bshell/status.h>
#include <fx/collections/array.h>
#include <fx/reflection/function.h>
#include <fx/string.h>
#include <fx/vector.h>

struct bshell_pipeline_p {
	bshell_pipeline *p_self;
	fx_array *p_in, *p_out;
	FX_VECTOR_DECLARE(bshell_cmdcall *, p_cmds);
};

static void pipeline_init(fx_object *obj, void *priv)
{
	struct bshell_pipeline_p *pipeline = priv;
	pipeline->p_self = obj;
	pipeline->p_in = fx_array_create();
	pipeline->p_out = fx_array_create();
}

static void pipeline_fini(fx_object *obj, void *priv)
{
	struct bshell_pipeline_p *pipeline = priv;
	for (size_t i = 0; i < pipeline->p_cmds.count; i++) {
		bshell_cmdcall_unref(pipeline->p_cmds.items[i]);
	}

	if (pipeline->p_in) {
		fx_array_unref(pipeline->p_in);
		pipeline->p_in = NULL;
	}

	if (pipeline->p_out) {
		fx_array_unref(pipeline->p_out);
		pipeline->p_out = NULL;
	}
}

static enum bshell_status write_value(
	struct bshell_pipeline_p *pipeline,
	fx_array *dest,
	fx_value value,
	bool enumerate)
{
	if (!enumerate) {
		fx_array_push_back(dest, value);
		return BSHELL_SUCCESS;
	}

	fx_object *container = NULL;
	fx_value_get_object(&value, &container);
	if (!container) {
		fx_array_push_back(dest, value);
		return BSHELL_SUCCESS;
	}

	const fx_iterator *it = fx_iterator_begin(container);
	fx_foreach(v, it)
	{
		fx_array_push_back(dest, fx_value_ref_copy_return(v));
	}
	fx_iterator_unref(it);

	return BSHELL_SUCCESS;
}

static enum bshell_status pipeline_add_input_value(
	struct bshell_pipeline_p *pipeline,
	fx_value value,
	bool enumerate)
{
	return write_value(pipeline, pipeline->p_in, value, enumerate);
}

static enum bshell_status pipeline_add_cmdcall(
	struct bshell_pipeline_p *pipeline,
	bshell_cmdcall *cmdcall)
{
	bshell_cmdcall **slot = fx_vector_emplace_back(pipeline->p_cmds, NULL);
	*slot = bshell_cmdcall_ref(cmdcall);
	return BSHELL_SUCCESS;
}

static enum bshell_status pipeline_execute_single(
	struct bshell_pipeline_p *pipeline)
{
	return BSHELL_ERR_NOT_SUPPORTED;
}

static fx_value pipeline_read_value(struct bshell_pipeline_p *pipeline)
{
	return fx_array_pop_front(pipeline->p_in);
}

static enum bshell_status pipeline_write_value(
	struct bshell_pipeline_p *pipeline,
	fx_value value,
	bool enumerate)
{
	return write_value(pipeline, pipeline->p_out, value, enumerate);
}

static enum bshell_status pipeline_pump_record(
	struct bshell_pipeline_p *pipeline,
	struct bshell_runtime *rt,
	fx_value *result,
	int *out_end_of_data)
{
	enum bshell_status status = BSHELL_SUCCESS;
	if (fx_array_get_size(pipeline->p_out)) {
		fx_value out = fx_array_pop_front(pipeline->p_out);
		*result = out;
		return BSHELL_SUCCESS;
	}

	size_t i = 0;
	for (i = 0; i < pipeline->p_cmds.count;) {
		status = bshell_cmdcall_process_record(
			pipeline->p_cmds.items[i],
			pipeline->p_self,
			rt);

		if (status != BSHELL_SUCCESS) {
			break;
		}

		bool input_available = fx_array_get_size(pipeline->p_in) != 0;
		bool output_available = fx_array_get_size(pipeline->p_out) != 0;

		if (input_available) {
			/* repeat this pipeline stage until all available input
			 * records are consumed */
			continue;
		}

		if (!output_available) {
			/* no output produced, and no more input available */
			break;
		}

		if (i < pipeline->p_cmds.count - 1) {
			fx_array *tmp = pipeline->p_in;
			pipeline->p_in = pipeline->p_out;
			pipeline->p_out = tmp;
		}
		i++;
	}

	if (fx_array_get_size(pipeline->p_out)) {
		fx_value out = fx_array_pop_front(pipeline->p_out);
		*result = out;
		return BSHELL_SUCCESS;
	} else if (i == 0) {
		*result = FX_VALUE_EMPTY;
		*out_end_of_data = 1;
	}

#if 0
	size_t i = 0;
	for (i = 0; i < pipeline->p_cmds.count; i++) {
		status = bshell_command_process_record(
			pipeline->p_cmds.items[i],
			rt);

		if (status != BSHELL_SUCCESS) {
			break;
		}

		fx_array *tmp = pipeline->p_in;
		pipeline->p_in = pipeline->p_out;
		pipeline->p_out = tmp;

		if (fx_array_size(pipeline->p_in)) {
			break;
		}
	}

	if (in->v_type) {
		fx_value_copy(result, in);
	} else if (i == 0) {
		*out_end_of_data = 1;
	}

	fx_value_unset(in);
	fx_value_unset(out);
#endif
	return status;
}

enum bshell_status bshell_pipeline_add_input_value(
	bshell_pipeline *pipeline,
	fx_value value,
	bool enumerate)
{
	FX_CLASS_DISPATCH_STATIC(
		BSHELL_TYPE_PIPELINE,
		pipeline_add_input_value,
		pipeline,
		value,
		enumerate);
}

enum bshell_status bshell_pipeline_add_cmdcall(
	bshell_pipeline *pipeline,
	bshell_cmdcall *cmd)
{
	FX_CLASS_DISPATCH_STATIC(
		BSHELL_TYPE_PIPELINE,
		pipeline_add_cmdcall,
		pipeline,
		cmd);
}

enum bshell_status bshell_pipeline_execute_single(bshell_pipeline *pipeline)
{
	FX_CLASS_DISPATCH_STATIC_0(
		BSHELL_TYPE_PIPELINE,
		pipeline_execute_single,
		pipeline);
}

enum bshell_status bshell_pipeline_pump_record(
	bshell_pipeline *pipeline,
	struct bshell_runtime *rt,
	fx_value *out,
	int *out_end_of_data)
{
	FX_CLASS_DISPATCH_STATIC(
		BSHELL_TYPE_PIPELINE,
		pipeline_pump_record,
		pipeline,
		rt,
		out,
		out_end_of_data);
}

fx_value bshell_pipeline_read_value(bshell_pipeline *pipeline)
{
	FX_CLASS_DISPATCH_STATIC_0(
		BSHELL_TYPE_PIPELINE,
		pipeline_read_value,
		pipeline);
}

enum bshell_status bshell_pipeline_write_value(
	bshell_pipeline *pipeline,
	fx_value val,
	bool enumerate)
{
	FX_CLASS_DISPATCH_STATIC(
		BSHELL_TYPE_PIPELINE,
		pipeline_write_value,
		pipeline,
		val,
		enumerate);
}

FX_TYPE_CLASS_BEGIN(bshell_pipeline)
	FX_TYPE_VTABLE_INTERFACE_BEGIN(fx_object, FX_TYPE_OBJECT)
		FX_INTERFACE_ENTRY(to_string) = NULL;
	FX_TYPE_VTABLE_INTERFACE_END(fx_object, FX_TYPE_OBJECT)

	FX_TYPE_CONSTRUCTOR("create", bshell_pipeline_create, 0, FX_TYPE_VOID);
FX_TYPE_CLASS_END(bshell_pipeline)

FX_TYPE_DEFINITION_BEGIN(bshell_pipeline)
	FX_TYPE_ID(0x64159f21, 0xb1dd, 0x4838, 0xba62, 0x69fa65cc01d3);
	FX_TYPE_NAME("bshell.pipeline");
	FX_TYPE_CLASS(bshell_pipeline_class);
	FX_TYPE_INSTANCE_PRIVATE(struct bshell_pipeline_p);
	FX_TYPE_INSTANCE_INIT(pipeline_init);
	FX_TYPE_INSTANCE_FINI(pipeline_fini);
FX_TYPE_DEFINITION_END(bshell_pipeline)