The C API supports setting the log level. (#158)

### Motivation
#137

### Modifications
-  Add interface `pulsar_client_configuration_set_logger_and_level` to support set the log level.

Author: shibd

examples/CMakeLists.txt

@@ -68,6 +68,10 @@ set(SAMPLE_KEY_VALUE_SCHEMA_PRODUCER
         SampleKeyValueSchemaProducer.cc
 )
 
+set(SAMPLE_CUSTOM_LOGGER_CAPI
+        SampleCustomLoggerCApi.c
+        )
+
 add_executable(SampleAsyncProducer                    ${SAMPLE_ASYNC_PRODUCER_SOURCES})
 add_executable(SampleConsumer                         ${SAMPLE_CONSUMER_SOURCES})
 add_executable(SampleConsumerListener                 ${SAMPLE_CONSUMER_LISTENER_SOURCES})
@@ -80,6 +84,7 @@ add_executable(SampleConsumerListenerCApi             ${SAMPLE_CONSUMER_LISTENER
 add_executable(SampleReaderCApi                       ${SAMPLE_READER_C_SOURCES})
 add_executable(SampleKeyValueSchemaConsumer           ${SAMPLE_KEY_VALUE_SCHEMA_CONSUMER})
 add_executable(SampleKeyValueSchemaProducer           ${SAMPLE_KEY_VALUE_SCHEMA_PRODUCER})
+add_executable(SampleCustomLoggerCApi                 ${SAMPLE_CUSTOM_LOGGER_CAPI})
 
 target_link_libraries(SampleAsyncProducer              ${CLIENT_LIBS} pulsarShared)
 target_link_libraries(SampleConsumer                   ${CLIENT_LIBS} pulsarShared)
@@ -93,3 +98,4 @@ target_link_libraries(SampleConsumerListenerCApi       ${CLIENT_LIBS} pulsarShar
 target_link_libraries(SampleReaderCApi                 ${CLIENT_LIBS} pulsarShared)
 target_link_libraries(SampleKeyValueSchemaConsumer     ${CLIENT_LIBS} pulsarShared)
 target_link_libraries(SampleKeyValueSchemaProducer     ${CLIENT_LIBS} pulsarShared)
+target_link_libraries(SampleCustomLoggerCApi           ${CLIENT_LIBS} pulsarShared)

examples/SampleCustomLoggerCApi.c

@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#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);
+    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) {
+    char *time_str = current_time();
+    printf("[%s] [%u] [%s] [%d] [%s] \n", time_str, level, file, line, message);
+    free(time_str);
+}
+
+int main() {
+    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_memory_limit(conf, 64 * 1024 * 1024);
+    pulsar_client_t *client = pulsar_client_create("pulsar://localhost:6650", conf);
+
+    pulsar_producer_configuration_t *producer_conf = pulsar_producer_configuration_create();
+    pulsar_producer_configuration_set_batching_enabled(producer_conf, 1);
+    pulsar_producer_t *producer;
+
+    pulsar_result err = pulsar_client_create_producer(client, "my-topic", producer_conf, &producer);
+    if (err != pulsar_result_Ok) {
+        printf("Failed to create producer: %s\n", pulsar_result_str(err));
+        return 1;
+    }
+
+    for (int i = 0; i < 10; i++) {
+        const char *data = "my-content";
+        pulsar_message_t *message = pulsar_message_create();
+        pulsar_message_set_content(message, data, strlen(data));
+
+        err = pulsar_producer_send(producer, message);
+        if (err == pulsar_result_Ok) {
+            printf("Sent message %d\n", i);
+        } else {
+            printf("Failed to publish message: %s\n", pulsar_result_str(err));
+            return 1;
+        }
+
+        pulsar_message_free(message);
+    }
+
+    // Cleanup
+    pulsar_producer_close(producer);
+    pulsar_producer_free(producer);
+    pulsar_producer_configuration_free(producer_conf);
+
+    pulsar_client_close(client);
+    pulsar_client_free(client);
+    pulsar_client_configuration_free(conf);
+}

include/pulsar/c/client_configuration.h

@@ -134,6 +134,10 @@ 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_use_tls(pulsar_client_configuration_t *conf, int useTls);
 
 PULSAR_PUBLIC int pulsar_client_configuration_is_use_tls(pulsar_client_configuration_t *conf);

lib/c/c_ClientConfiguration.cc

@@ -72,13 +72,14 @@ 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, void *ctx)
-        : file_(file), logger_(logger), ctx_(ctx) {}
+    PulsarCLogger(const std::string &file, pulsar_logger logger, pulsar_logger_level_t level, void *ctx)
+        : file_(file), logger_(logger), level_(level), ctx_(ctx) {}
 
-    bool isEnabled(Level level) { return level >= pulsar::Logger::LEVEL_INFO; }
+    bool isEnabled(Level level) { return (pulsar_logger_level_t)level >= level_; }
 
     void log(Level level, int line, const std::string &message) {
         logger_((pulsar_logger_level_t)level, file_.c_str(), line, message.c_str(), ctx_);
@@ -87,19 +88,27 @@ class PulsarCLogger : public pulsar::Logger {
 
 class PulsarCLoggerFactory : public pulsar::LoggerFactory {
     pulsar_logger logger_;
+    pulsar_logger_level_t level_;
     void *ctx_;
 
    public:
-    PulsarCLoggerFactory(pulsar_logger logger, void *ctx) : logger_(logger), ctx_(ctx) {}
+    PulsarCLoggerFactory(pulsar_logger logger, pulsar_logger_level_t level, void *ctx)
+        : logger_(logger), level_(level), ctx_(ctx) {}
 
     pulsar::Logger *getLogger(const std::string &fileName) {
-        return new PulsarCLogger(fileName, logger_, ctx_);
+        return new PulsarCLogger(fileName, logger_, level_, ctx_);
     }
 };
 
 void pulsar_client_configuration_set_logger(pulsar_client_configuration_t *conf, pulsar_logger logger,
                                             void *ctx) {
-    conf->conf.setLogger(new PulsarCLoggerFactory(logger, ctx));
+    conf->conf.setLogger(new PulsarCLoggerFactory(logger, pulsar_logger_level_t::pulsar_INFO, ctx));
+}
+
+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_use_tls(pulsar_client_configuration_t *conf, int useTls) {