proxy: go: add context for input methods

The way to add context to input methods is completely identical
to the approach used for output methods.

Without this, the callback can only perform hardwired tasks,
and the configuration values can not be accessed. As a result,
the input becomes extremely constrained on what it can do.

Signed-off-by: milkrage <[email protected]>

Author: milkrage

src/proxy/go/go.c

@@ -184,6 +184,7 @@ int proxy_go_input_register(struct flb_plugin_proxy *proxy,
      *
      * - FLBPluginInit
      * - FLBPluginInputCallback
+     * - FLBPluginInputCallbackCtx
      * - FLBPluginExit
      *
      * note: registration callback FLBPluginRegister() is resolved by the
@@ -198,7 +199,9 @@ int proxy_go_input_register(struct flb_plugin_proxy *proxy,
     }
 
     plugin->cb_collect = flb_plugin_proxy_symbol(proxy, "FLBPluginInputCallback");
+    plugin->cb_collect_ctx = flb_plugin_proxy_symbol(proxy, "FLBPluginInputCallbackCtx");
     plugin->cb_cleanup = flb_plugin_proxy_symbol(proxy, "FLBPluginInputCleanupCallback");
+    plugin->cb_cleanup_ctx = flb_plugin_proxy_symbol(proxy, "FLBPluginInputCleanupCallbackCtx");
     plugin->cb_exit  = flb_plugin_proxy_symbol(proxy, "FLBPluginExit");
     plugin->name     = flb_strdup(def->name);
 
@@ -231,27 +234,35 @@ int proxy_go_input_init(struct flb_plugin_proxy *proxy)
     return ret;
 }
 
-int proxy_go_input_collect(struct flb_plugin_proxy *ctx,
+int proxy_go_input_collect(struct flb_plugin_input_proxy_context *ctx,
                            void **collected_data, size_t *len)
 {
-    int ret;
+    int ret = -1;
     void *data = NULL;
-    struct flbgo_input_plugin *plugin = ctx->data;
+    struct flbgo_input_plugin *plugin = ctx->proxy->data;
 
-    ret = plugin->cb_collect(&data, len);
+    if (plugin->cb_collect_ctx) {
+        ret = plugin->cb_collect_ctx(ctx->remote_context, &data, len);
+    }
+    else if (plugin->cb_collect) {
+        ret = plugin->cb_collect(&data, len);
+    }
 
     *collected_data = data;
 
     return ret;
 }
 
-int proxy_go_input_cleanup(struct flb_plugin_proxy *ctx,
+int proxy_go_input_cleanup(struct flb_plugin_input_proxy_context *ctx,
                            void *allocated_data)
 {
     int ret = 0;
-    struct flbgo_input_plugin *plugin = ctx->data;
+    struct flbgo_input_plugin *plugin = ctx->proxy->data;
 
-    if (plugin->cb_cleanup) {
+    if (plugin->cb_cleanup_ctx) {
+        ret = plugin->cb_cleanup_ctx(ctx->remote_context, allocated_data);
+    }
+    else if (plugin->cb_cleanup) {
         ret = plugin->cb_cleanup(allocated_data);
     }
     else {

src/proxy/go/go.h

@@ -40,11 +40,13 @@ struct flbgo_input_plugin {
     char *name;
     void *api;
     void *i_ins;
-    struct flb_plugin_proxy_context *context;
+    struct flb_plugin_input_proxy_context *context;
 
     int (*cb_init)();
     int (*cb_collect)(void **, size_t *);
+    int (*cb_collect_ctx)(void *, void **, size_t *);
     int (*cb_cleanup)(void *);
+    int (*cb_cleanup_ctx)(void *, void *);
     int (*cb_exit)();
 };
 
@@ -63,9 +65,9 @@ int proxy_go_input_register(struct flb_plugin_proxy *proxy,
                             struct flb_plugin_proxy_def *def);
 
 int proxy_go_input_init(struct flb_plugin_proxy *proxy);
-int proxy_go_input_collect(struct flb_plugin_proxy *ctx,
+int proxy_go_input_collect(struct flb_plugin_input_proxy_context *ctx,
                            void **collected_data, size_t *len);
-int proxy_go_input_cleanup(struct flb_plugin_proxy *ctx,
+int proxy_go_input_cleanup(struct flb_plugin_input_proxy_context *ctx,
                            void *allocated_data);
 int proxy_go_input_destroy(struct flb_plugin_input_proxy_context *ctx);
 void proxy_go_input_unregister(void *data);