#include #include #include #include #include #include #include struct bshell_pipeline_p { bool p_input_values_provided; 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]); } fx_vector_destroy(pipeline->p_cmds, NULL); 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) { pipeline->p_input_values_provided = true; 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;) { enum bshell_command_position position; if (pipeline->p_cmds.count == 1) { position = pipeline->p_input_values_provided ? BSHELL_COMMAND_POSITION_END : BSHELL_COMMAND_POSITION_SINGLE; } else if (i == 0) { position = pipeline->p_input_values_provided ? BSHELL_COMMAND_POSITION_END : BSHELL_COMMAND_POSITION_BEGINNING; } else if (i == pipeline->p_cmds.count - 1) { position = BSHELL_COMMAND_POSITION_END; } else { position = BSHELL_COMMAND_POSITION_MIDDLE; } status = bshell_cmdcall_process_record( pipeline->p_cmds.items[i], pipeline->p_self, rt, position); 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)