out_kafka_rest: use flb_msgpack_raw_to_json_sds() instead of old ...raw_to_json_str()
#include <fluent-bit/flb_utils.h>
#include <fluent-bit/flb_time.h>
#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_sds.h>
#include <fluent-bit/flb_http_client.h>
#include <msgpack.h>
* Convert the internal Fluent Bit data representation to the required
* one by Kafka REST Proxy.
*/
-static char *kafka_rest_format(const void *data, size_t bytes,
- const char *tag, int tag_len, size_t *out_size,
+static flb_sds_t kafka_rest_format(const void *data, size_t bytes,
+ const char *tag, int tag_len,
+ size_t *out_size,
struct flb_kafka_rest *ctx)
{
int i;
- int ret;
int len;
int arr_size = 0;
int map_size;
size_t s;
size_t off = 0;
+ flb_sds_t out_buf;
char time_formatted[256];
msgpack_unpacked result;
msgpack_object root;
@@ -51,8 +53,6 @@ static char *kafka_rest_format(const void *data, size_t bytes,
msgpack_object *obj;
msgpack_object key;
msgpack_object val;
- char *json_buf;
- size_t json_size;
struct tm tm;
struct flb_time tms;
msgpack_sbuffer mp_sbuf;
@@ -157,15 +157,14 @@ static char *kafka_rest_format(const void *data, size_t bytes,
msgpack_unpacked_destroy(&result);
/* Convert to JSON */
- ret = flb_msgpack_raw_to_json_str(mp_sbuf.data, mp_sbuf.size,
- &json_buf, &json_size);
+ out_buf = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
msgpack_sbuffer_destroy(&mp_sbuf);
- if (ret != 0) {
+ if (!out_buf) {
return NULL;
}
- *out_size = json_size;
- return json_buf;
+ *out_size = flb_sds_len(out_buf);
+ return out_buf;
}
static int cb_kafka_init(struct flb_output_instance *ins,
@@ -197,7 +196,7 @@ static void cb_kafka_flush(const void *data, size_t bytes,
struct flb_config *config)
{
int ret;
- char *js;
+ flb_sds_t js;
size_t js_size;
size_t b_sent;
struct flb_http_client *c;
@@ -259,14 +258,14 @@ static void cb_kafka_flush(const void *data, size_t bytes,
/* Cleanup */
flb_http_client_destroy(c);
- flb_free(js);
+ flb_sds_destroy(js);
flb_upstream_conn_release(u_conn);
FLB_OUTPUT_RETURN(FLB_OK);
/* Issue a retry */
retry:
flb_http_client_destroy(c);
- flb_free(js);
+ flb_sds_destroy(js);
flb_upstream_conn_release(u_conn);
FLB_OUTPUT_RETURN(FLB_RETRY);
}
