[feat] Add pulsar_logger_t as the configurable C logger (#162)

### Motivation

The current `pulsar_client_configuration_set_logger` API can only
configure the `Logger::log` method, but the` Logger::isEnabled` method
cannot be configured via C API.

#158 added a
`pulsar_client_configuration_set_logger_and_level` function to configure
a log level, but it's not flexible. For example, the log level might be
modified dynamically (though it's a complicated case).

### Modifications

Add a `pulsar_logger_t` struct and the related
`pulsar_client_configuration_set_logger_t` function to configure it as
the C logger API. The `is_enabled` and `log` fields of the struct are
the responding methods of the `isEnabled` and `log` methods in C++
`Logger`.

Then add a `LogContext` example in `SampleCustomLoggerCApi.c` to print
logs to a file or standard output.

Eliminate the `pulsar_client_configuration_set_logger_and_level`
function.

examples/SampleCustomLoggerCApi.c

@@ -17,31 +17,60 @@
  * under the License.
  */
 
+#include <ctype.h>
 #include <pulsar/c/client.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <time.h>
 
 char *current_time() {
-    char *time_str = malloc(128);
+    char *time_str = (char *)malloc(128);
     struct tm *p;
     time_t now = time(0);
     p = gmtime(&now);
     strftime(time_str, 128, "%Y-%m-%d %H:%M:%S", p);
     return time_str;
 }
 
-void custom_logger(pulsar_logger_level_t level, const char *file, int line, const char *message, void *ctx) {
+typedef struct LogContext {
+    FILE *file;
+    pulsar_logger_level_t level;
+} LogContext;
+
+void log_context_init(LogContext *ctx, const char *level, const char *filename);
+void log_context_destroy(LogContext *ctx);
+
+bool is_enabled(pulsar_logger_level_t level, void *ctx) { return level >= ((LogContext *)ctx)->level; }
+
+void log_func(pulsar_logger_level_t level, const char *file, int line, const char *message, void *ctx) {
     char *time_str = current_time();
-    printf("[%s] [%u] [%s] [%d] [%s] \n", time_str, level, file, line, message);
+    fprintf(((LogContext *)ctx)->file, "[%s] [%u] [%s] [%d] [%s] \n", time_str, level, file, line, message);
     free(time_str);
 }
 
-int main() {
+int main(int argc, char *argv[]) {
+    if (argc < 2) {
+        fprintf(stderr,
+                "Usage: %s log-level <filename>\n\n"
+                "  log-level could be DEBUG, INFO, WARN or ERROR\n"
+                "  If filename is specified, logs will be printed into the given file.\n"
+                "  Otherwise, logs will be printed into the standard output.\n",
+                argv[0]);
+        return 1;
+    }
+
+    LogContext ctx;
+    log_context_init(&ctx, argv[1], (argc > 2) ? argv[2] : NULL);
+
+    pulsar_logger_t logger;
+    logger.ctx = &ctx;
+    logger.is_enabled = &is_enabled;
+    logger.log = &log_func;
+
     pulsar_client_configuration_t *conf = pulsar_client_configuration_create();
 
-    pulsar_client_configuration_set_logger_and_level(conf, custom_logger, pulsar_DEBUG, NULL);
+    pulsar_client_configuration_set_logger_t(conf, logger);
     pulsar_client_configuration_set_memory_limit(conf, 64 * 1024 * 1024);
     pulsar_client_t *client = pulsar_client_create("pulsar://localhost:6650", conf);
 
@@ -79,4 +108,46 @@ int main() {
     pulsar_client_close(client);
     pulsar_client_free(client);
     pulsar_client_configuration_free(conf);
+    log_context_destroy(&ctx);
+}
+
+static bool str_equal_ignore_case(const char *lhs, const char *rhs) {
+    int length = strlen(lhs);
+    for (int i = 0; i < length; i++) {
+        if (lhs[i] != rhs[i]) {
+            return false;
+        }
+    }
+    return true;
+}
+
+void log_context_init(LogContext *ctx, const char *level, const char *filename) {
+    if (str_equal_ignore_case(level, "debug")) {
+        ctx->level = pulsar_DEBUG;
+    } else if (str_equal_ignore_case(level, "info")) {
+        ctx->level = pulsar_INFO;
+    } else if (str_equal_ignore_case(level, "warn")) {
+        ctx->level = pulsar_WARN;
+    } else if (str_equal_ignore_case(level, "error")) {
+        ctx->level = pulsar_ERROR;
+    } else {
+        fprintf(stderr, "Unknown log level: %s\n", level);
+        exit(1);
+    }
+
+    if (filename) {
+        ctx->file = fopen(filename, "w+");
+        if (!ctx->file) {
+            fprintf(stderr, "Failed to open %s\n", filename);
+            exit(2);
+        }
+    } else {
+        ctx->file = stdout;
+    }
+}
+
+void log_context_destroy(LogContext *ctx) {
+    if (ctx && ctx->file && ctx->file != stdout) {
+        fclose(ctx->file);
+    }
 }

include/pulsar/c/client_configuration.h

@@ -20,6 +20,7 @@
 #pragma once
 
 #include <pulsar/defines.h>
+#include <stdbool.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -36,6 +37,15 @@ typedef enum
 typedef void (*pulsar_logger)(pulsar_logger_level_t level, const char *file, int line, const char *message,
                               void *ctx);
 
+typedef struct pulsar_logger_t {
+    // The context that will be passed into `is_enabled` and `log` as the last argument
+    void *ctx;
+    // Whether to log for the given log level
+    bool (*is_enabled)(pulsar_logger_level_t level, void *ctx);
+    // How to log the message
+    pulsar_logger log;
+} pulsar_logger_t;
+
 typedef struct _pulsar_client_configuration pulsar_client_configuration_t;
 typedef struct _pulsar_authentication pulsar_authentication_t;
 
@@ -134,9 +144,8 @@ PULSAR_PUBLIC int pulsar_client_configuration_get_concurrent_lookup_request(
 PULSAR_PUBLIC void pulsar_client_configuration_set_logger(pulsar_client_configuration_t *conf,
                                                           pulsar_logger logger, void *ctx);
 
-PULSAR_PUBLIC void pulsar_client_configuration_set_logger_and_level(pulsar_client_configuration_t *conf,
-                                                                    pulsar_logger logger,
-                                                                    pulsar_logger_level_t level, void *ctx);
+PULSAR_PUBLIC void pulsar_client_configuration_set_logger_t(pulsar_client_configuration_t *conf,
+                                                            pulsar_logger_t logger);
 
 PULSAR_PUBLIC void pulsar_client_configuration_set_use_tls(pulsar_client_configuration_t *conf, int useTls);
 

lib/c/c_ClientConfiguration.cc

@@ -70,45 +70,49 @@ int pulsar_client_configuration_get_concurrent_lookup_request(pulsar_client_conf
 }
 
 class PulsarCLogger : public pulsar::Logger {
-    std::string file_;
-    pulsar_logger logger_;
-    pulsar_logger_level_t level_;
-    void *ctx_;
-
    public:
-    PulsarCLogger(const std::string &file, pulsar_logger logger, pulsar_logger_level_t level, void *ctx)
-        : file_(file), logger_(logger), level_(level), ctx_(ctx) {}
+    PulsarCLogger(pulsar_logger_t logger, const std::string &fileName)
+        : logger_(logger), fileName_(fileName) {}
 
-    bool isEnabled(Level level) { return (pulsar_logger_level_t)level >= level_; }
+    bool isEnabled(Level level) override {
+        return logger_.is_enabled(static_cast<pulsar_logger_level_t>(level), logger_.ctx);
+    }
 
-    void log(Level level, int line, const std::string &message) {
-        logger_((pulsar_logger_level_t)level, file_.c_str(), line, message.c_str(), ctx_);
+    void log(Level level, int line, const std::string &message) override {
+        logger_.log(static_cast<pulsar_logger_level_t>(level), fileName_.c_str(), line, message.c_str(),
+                    logger_.ctx);
     }
+
+   private:
+    const pulsar_logger_t logger_;
+    const std::string fileName_;
 };
 
 class PulsarCLoggerFactory : public pulsar::LoggerFactory {
-    pulsar_logger logger_;
-    pulsar_logger_level_t level_;
-    void *ctx_;
-
    public:
-    PulsarCLoggerFactory(pulsar_logger logger, pulsar_logger_level_t level, void *ctx)
-        : logger_(logger), level_(level), ctx_(ctx) {}
+    PulsarCLoggerFactory(pulsar_logger_t logger) : logger_(logger) {}
 
-    pulsar::Logger *getLogger(const std::string &fileName) {
-        return new PulsarCLogger(fileName, logger_, level_, ctx_);
+    pulsar::Logger *getLogger(const std::string &fileName) override {
+        return new PulsarCLogger(logger_, fileName);
     }
+
+   private:
+    const pulsar_logger_t logger_;
 };
 
-void pulsar_client_configuration_set_logger(pulsar_client_configuration_t *conf, pulsar_logger logger,
-                                            void *ctx) {
-    conf->conf.setLogger(new PulsarCLoggerFactory(logger, pulsar_logger_level_t::pulsar_INFO, ctx));
+void pulsar_client_configuration_set_logger(pulsar_client_configuration_t *conf,
+                                            pulsar_logger logger_function, void *ctx) {
+    pulsar_logger_t logger;
+    logger.ctx = ctx;
+    logger.is_enabled = [](pulsar_logger_level_t level, void *ctx) {
+        return level >= pulsar_logger_level_t::pulsar_INFO;
+    };
+    logger.log = logger_function;
+    conf->conf.setLogger(new PulsarCLoggerFactory(logger));
 }
 
-void pulsar_client_configuration_set_logger_and_level(pulsar_client_configuration_t *conf,
-                                                      pulsar_logger logger, pulsar_logger_level_t level,
-                                                      void *ctx) {
-    conf->conf.setLogger(new PulsarCLoggerFactory(logger, level, ctx));
+void pulsar_client_configuration_set_logger_t(pulsar_client_configuration_t *conf, pulsar_logger_t logger) {
+    conf->conf.setLogger(new PulsarCLoggerFactory(logger));
 }
 
 void pulsar_client_configuration_set_use_tls(pulsar_client_configuration_t *conf, int useTls) {