tests: runtime: in_tcp: add test case for issue 5336
@@ -60,6 +60,28 @@ static void clear_output_num()
set_output_num(0);
}
+static int cb_count_msgpack(void *record, size_t size, void *data)
+{
+ msgpack_unpacked result;
+ size_t off = 0;
+
+ if (!TEST_CHECK(data != NULL)) {
+ flb_error("data is NULL");
+ }
+
+ /* Iterate each item array and apply rules */
+ msgpack_unpacked_init(&result);
+ while (msgpack_unpack_next(&result, record, size, &off) == MSGPACK_UNPACK_SUCCESS) {
+ pthread_mutex_lock(&result_mutex);
+ num_output++;
+ pthread_mutex_unlock(&result_mutex);
+ }
+ msgpack_unpacked_destroy(&result);
+
+ flb_free(record);
+ return 0;
+}
+
/* Callback to check expected results */
static int cb_check_result_json(void *record, size_t size, void *data)
{
@@ -354,10 +376,77 @@ void flb_test_format_none_separator()
test_ctx_destroy(ctx);
}
+/*
+ * Ingest 64k records.
+ * https://github.com/fluent/fluent-bit/issues/5336
+ */
+void flb_test_issue_5336()
+{
+ struct flb_lib_out_cb cb_data;
+ struct test_ctx *ctx;
+ flb_sockfd_t fd;
+ int ret;
+ int num;
+ ssize_t w_size;
+ int not_used;
+ int i;
+ int count = 65535;
+
+ char *buf = "{\"test\":\"msg\"}";
+ size_t size = strlen(buf);
+
+ clear_output_num();
+
+ cb_data.cb = cb_count_msgpack;
+ cb_data.data = &not_used;
+
+ ctx = test_ctx_create(&cb_data);
+ if (!TEST_CHECK(ctx != NULL)) {
+ TEST_MSG("test_ctx_create failed");
+ exit(EXIT_FAILURE);
+ }
+
+ ret = flb_output_set(ctx->flb, ctx->o_ffd,
+ "match", "*",
+ NULL);
+ TEST_CHECK(ret == 0);
+
+ /* Start the engine */
+ ret = flb_start(ctx->flb);
+ TEST_CHECK(ret == 0);
+
+ /* use default host/port */
+ fd = connect_tcp(NULL, -1);
+ if (!TEST_CHECK(fd >= 0)) {
+ exit(EXIT_FAILURE);
+ }
+
+ for (i=0; i<count; i++) {
+ w_size = send(fd, buf, size, 0);
+ if (!TEST_CHECK(w_size == size)) {
+ TEST_MSG("failed to send, count=%d errno=%d",i, errno);
+ flb_socket_close(fd);
+ exit(EXIT_FAILURE);
+ }
+ }
+
+ /* waiting to flush */
+ flb_time_msleep(2500);
+
+ num = get_output_num();
+ if (!TEST_CHECK(num == count)) {
+ TEST_MSG("got %d, expected: %d", num, count);
+ }
+
+ flb_socket_close(fd);
+ test_ctx_destroy(ctx);
+}
+
TEST_LIST = {
{"tcp", flb_test_tcp},
{"format_none", flb_test_format_none},
{"format_none_separator", flb_test_format_none_separator},
+ {"65535_records_issue_5336", flb_test_issue_5336},
{NULL, NULL}
};
