task: process direct routes if they are set
@@ -347,8 +347,10 @@ struct flb_task *flb_task_create(uint64_t ref_id,
struct flb_task *task;
struct flb_event_chunk *evc;
struct flb_task_route *route;
+ struct flb_router_path *route_path;
struct flb_output_instance *o_ins;
struct flb_input_chunk *task_ic;
+ struct mk_list *i_head;
struct mk_list *o_head;
/* No error status */
@@ -389,6 +391,27 @@ struct flb_task *flb_task_create(uint64_t ref_id,
task->records = ((struct flb_input_chunk *) ic)->total_records;
#endif
+ /* Direct connects betweek input <> outputs (API based) */
+ if (mk_list_size(&i_ins->routes_direct) > 0) {
+ mk_list_foreach(i_head, &i_ins->routes_direct) {
+ route_path = mk_list_entry(i_head, struct flb_router_path, _head);
+ o_ins = route_path->ins;
+
+ route = flb_malloc(sizeof(struct flb_task_route));
+ if (!route) {
+ flb_errno();
+ task->event_chunk->data = NULL;
+ flb_task_destroy(task, FLB_TRUE);
+ return NULL;
+ }
+
+ route->out = o_ins;
+ mk_list_add(&route->_head, &task->routes);
+ }
+ flb_debug("[task] created direct task=%p id=%i OK", task, task->id);
+ return task;
+ }
+
/* Find matching routes for the incoming task */
mk_list_foreach(o_head, &config->outputs) {
o_ins = mk_list_entry(o_head,
