input: fix handling of pause/resume events and collector status
@@ -644,7 +644,6 @@ int flb_input_collector_pause(int coll_id, struct flb_input_instance *in)
{
int ret;
struct flb_config *config;
- struct mk_event *event;
struct flb_input_collector *coll;
coll = get_collector(coll_id, in);
@@ -652,7 +651,6 @@ int flb_input_collector_pause(int coll_id, struct flb_input_instance *in)
return -1;
}
- event = &coll->event;
config = in->config;
if (coll->type == FLB_COLLECT_TIME) {
/*
@@ -661,19 +659,17 @@ int flb_input_collector_pause(int coll_id, struct flb_input_instance *in)
* one can be created.
*/
mk_event_timeout_destroy(config->evl, &coll->event);
- mk_event_del(config->evl, &coll->event);
close(coll->fd_timer);
coll->fd_timer = -1;
}
else if (coll->type & (FLB_COLLECT_FD_SERVER | FLB_COLLECT_FD_EVENT)) {
- ret = mk_event_add(config->evl,
- coll->fd_event,
- FLB_ENGINE_EV_CORE,
- MK_EVENT_IDLE, event);
+ ret = mk_event_del(config->evl, &coll->event);
if (ret != 0) {
flb_warn("[input] cannot disable event for %s", in->name);
+ return -1;
}
}
+
coll->running = FLB_FALSE;
return 0;
@@ -713,16 +709,22 @@ int flb_input_collector_resume(int coll_id, struct flb_input_instance *in)
coll->fd_timer = fd;
}
else if (coll->type & (FLB_COLLECT_FD_SERVER | FLB_COLLECT_FD_EVENT)) {
+ event->fd = coll->fd_event;
+ event->mask = MK_EVENT_EMPTY;
+ event->status = MK_EVENT_NONE;
+
ret = mk_event_add(config->evl,
coll->fd_event,
FLB_ENGINE_EV_CORE,
MK_EVENT_READ, event);
- if (ret != 0) {
- flb_warn("[input] cannot disable event for %s", in->name);
+ if (ret == -1) {
+ flb_error("[input] cannot disable/pause event for %s", in->name);
return -1;
}
}
+ coll->running = FLB_TRUE;
+
return 0;
}
@@ -977,6 +979,10 @@ int flb_input_collector_fd(flb_pipefd_t fd, struct flb_config *config)
return -1;
}
+ if (collector->running == FLB_FALSE) {
+ return -1;
+ }
+
/* Trigger the collector callback */
if (collector->instance->threaded == FLB_TRUE) {
th = flb_input_thread_collect(collector, config);
