out_es: sds: Implement es upstream with sds view

include/fluent-bit/flb_lib.h

@@ -23,6 +23,8 @@
 #include <fluent-bit/flb_macros.h>
 #include <fluent-bit/flb_config.h>
 
+struct flb_input_instance;
+
 /* Lib engine status */
 #define FLB_LIB_ERROR     -1
 #define FLB_LIB_NONE       0
@@ -70,6 +72,16 @@ FLB_EXPORT int flb_output_set_test(flb_ctx_t *ctx, int ffd, char *test_name,
                                                          void *, size_t, void *),
                                    void *out_callback_data,
                                    void *test_ctx);
+FLB_EXPORT int flb_output_set_test_with_ctx_callback(
+                                   flb_ctx_t *ctx, int ffd, char *test_name,
+                                   void (*out_callback) (void *, int, int,
+                                                         void *, size_t, void *),
+                                   void *out_callback_data,
+                                   void *test_ctx,
+                                   void *(*test_ctx_callback) (
+                                           struct flb_config *,
+                                           struct flb_input_instance *,
+                                           void *, void *));
 FLB_EXPORT int flb_output_set_callback(flb_ctx_t *ctx, int ffd, char *name,
                                        void (*cb)(char *, void *, void *));
 FLB_EXPORT int flb_output_set_http_test(flb_ctx_t *ctx, int ffd, char *test_name,

include/fluent-bit/flb_output.h

@@ -154,9 +154,24 @@ struct flb_test_out_formatter {
      */
     void *rt_data;
 
-    /* optional context for flush callback */
+    /* optional context for "flush context callback" */
     void *flush_ctx;
 
+    /*
+     * Callback
+     * =========
+     * Optional "flush context callback": it references the function that extracts
+     * optional flush context for "formatter callback".
+     */
+    void *(*flush_ctx_callback) (/* Fluent Bit context */
+                                 struct flb_config *,
+                                 /* plugin that ingested the records */
+                                 struct flb_input_instance *,
+                                 /* plugin instance context */
+                                 void *plugin_context,
+                                 /* context for "flush context callback" */
+                                 void *flush_ctx);
+
     /*
      * Callback
      * =========

include/fluent-bit/flb_sds.h

@@ -35,6 +35,12 @@
 
 typedef char *flb_sds_t;
 
+struct flb_sds_view {
+    const char *buf;
+    size_t len;
+};
+typedef struct flb_sds_view flb_sds_view_t;
+
 #pragma pack(push, 1)
 struct flb_sds {
     uint64_t len;        /* used */
@@ -95,8 +101,33 @@ static inline int flb_sds_casecmp(flb_sds_t s, const char *str, int len)
     return strncasecmp(s, str, len);
 }
 
+static inline flb_sds_view_t flb_sds_view_create(const char *str, size_t len)
+{
+    flb_sds_view_t view;
+
+    view.buf = str;
+    view.len = len;
+
+    return view;
+}
+
+static inline flb_sds_view_t flb_sds_view_create_from_sds(flb_sds_t s)
+{
+    return flb_sds_view_create(s, flb_sds_len(s));
+}
+
+static inline int flb_sds_view_is_empty(flb_sds_view_t view)
+{
+    if (view.len == 0) {
+        return FLB_TRUE;
+    }
+
+    return FLB_FALSE;
+}
+
 flb_sds_t flb_sds_create(const char *str);
 flb_sds_t flb_sds_create_len(const char *str, int len);
+flb_sds_t flb_sds_create_from_view(flb_sds_view_t view);
 flb_sds_t flb_sds_create_size(size_t size);
 int flb_sds_trim(flb_sds_t s);
 flb_sds_t flb_sds_cat(flb_sds_t s, const char *str, int len);