#include <stdio.h>
#include <string.h>
#include <librdkafka/rdkafka.h>

int main (int argc, char **argv) {
        rd_kafka_conf_t *conf;
        rd_kafka_t *rk;
        char features[256];
        size_t fsize = sizeof(features);
        char errstr[512];
        const char *exp_features[] = {
                "gzip",
                "snappy",
                "ssl",
                "sasl",
                "regex",
                "lz4",
                "sasl_gssapi",
                "sasl_plain",
                "sasl_scram",
                "plugins",
                "zstd",
                "sasl_oauthbearer",
                NULL,
        };
        const char **exp;
        int missing = 0;


        printf("librdkafka %s\n", rd_kafka_version_str());

        conf = rd_kafka_conf_new();
        if (rd_kafka_conf_get(conf, "builtin.features", features, &fsize) !=
            RD_KAFKA_CONF_OK) {
                fprintf(stderr, "conf_get failed\n");
                return 1;
        }

        printf("builtin.features %s\n", features);

        /* Verify that expected features are enabled. */
        for (exp = exp_features ; *exp ; exp++) {
                const char *t = features;
                size_t elen = strlen(*exp);
                int match = 0;

                while ((t = strstr(t, *exp))) {
                        if (t[elen] == ',' ||
                            t[elen] == '\0') {
                                match = 1;
                                break;
                        }
                        t += elen;
                }

                if (match)
                        continue;

                fprintf(stderr, "ERROR: feature %s not found\n", *exp);
                missing++;
        }

        if (rd_kafka_conf_set(conf, "security.protocol", "SASL_SSL",
                              errstr, sizeof(errstr)) ||
            rd_kafka_conf_set(conf, "sasl.mechanism", "PLAIN",
                              errstr, sizeof(errstr)) ||
            rd_kafka_conf_set(conf, "sasl.username", "username",
                              errstr, sizeof(errstr)) ||
            rd_kafka_conf_set(conf, "sasl.password", "password",
                              errstr, sizeof(errstr)) ||
            rd_kafka_conf_set(conf, "debug", "security",
                              errstr, sizeof(errstr))) {
                fprintf(stderr, "conf_set failed: %s\n", errstr);
                return 1;
        }

        rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
        if (!rk) {
                fprintf(stderr, "rd_kafka_new failed: %s\n", errstr);
                return 1;
        }

        printf("client name %s\n", rd_kafka_name(rk));

        rd_kafka_destroy(rk);

        return missing ? 1 : 0;
}