downstream: add flb_downstream_pause and flb_downstream_resume api

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

include/fluent-bit/flb_downstream.h

@@ -45,6 +45,9 @@ struct flb_downstream {
 
     /* this is a config map reference coming from the plugin net_setup field */
     struct flb_net_setup  *net_setup;
+
+    /* pause state */
+    int paused;
 };
 
 static inline int flb_downstream_is_shutting_down(struct flb_downstream *downstream)
@@ -76,6 +79,9 @@ int flb_downstream_set_property(struct flb_config *config,
 
 struct flb_connection *flb_downstream_conn_get(struct flb_downstream *stream);
 
+void flb_downstream_pause(struct flb_downstream *stream);
+void flb_downstream_resume(struct flb_downstream *stream);
+
 int flb_downstream_conn_release(struct flb_connection *connection);
 
 int flb_downstream_conn_pending_destroy_list(struct mk_list *list);

src/flb_downstream.c

@@ -283,6 +283,18 @@ struct flb_connection *flb_downstream_conn_get(struct flb_downstream *stream)
 
     transport = stream->base.transport;
 
+    if (stream->paused) {
+        if (transport != FLB_TRANSPORT_UDP &&
+            transport != FLB_TRANSPORT_UNIX_DGRAM) {
+            connection_fd = flb_net_accept(stream->server_fd);
+            if (connection_fd >= 0) {
+                flb_socket_close(connection_fd);
+            }
+        }
+
+        return NULL;
+    }
+
     if (transport == FLB_TRANSPORT_UDP ||
         transport == FLB_TRANSPORT_UNIX_DGRAM ) {
         if (stream->dgram_connection != NULL) {
@@ -357,6 +369,20 @@ struct flb_connection *flb_downstream_conn_get(struct flb_downstream *stream)
     return connection;
 }
 
+void flb_downstream_pause(struct flb_downstream *stream)
+{
+    if (stream) {
+        stream->paused = FLB_TRUE;
+    }
+}
+
+void flb_downstream_resume(struct flb_downstream *stream)
+{
+    if (stream) {
+        stream->paused = FLB_FALSE;
+    }
+}
+
 void flb_downstream_destroy(struct flb_downstream *stream)
 {
     struct flb_connection *connection;