Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions examples/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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})
Expand All @@ -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)
Expand All @@ -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)
82 changes: 82 additions & 0 deletions examples/SampleCustomLoggerCApi.c
Original file line number Diff line number Diff line change
@@ -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);
}
4 changes: 4 additions & 0 deletions include/pulsar/c/client_configuration.h
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
21 changes: 15 additions & 6 deletions lib/c/c_ClientConfiguration.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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_);
Expand All @@ -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) {
Expand Down