1 /* 2 * Copyright (c) 2014 Pierre-Yves Ritschard <pyr@spootnik.org> 3 * Copyright (c) 2013-2019 Balabit 4 * Copyright (c) 2019 Balazs Scheidler 5 * 6 * This program is free software; you can redistribute it and/or modify it 7 * under the terms of the GNU General Public License version 2 as published 8 * by the Free Software Foundation, or (at your option) any later version. 9 * 10 * This program is distributed in the hope that it will be useful, 11 * but WITHOUT ANY WARRANTY; without even the implied warranty of 12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 13 * GNU General Public License for more details. 14 * 15 * You should have received a copy of the GNU General Public License 16 * along with this program; if not, write to the Free Software 17 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA 18 * 19 * As an additional exemption you are allowed to compile & link against the 20 * OpenSSL libraries as published by the OpenSSL project. See the file 21 * COPYING for details. 22 * 23 */ 24 25 #ifndef KAFKA_H_INCLUDED 26 #define KAFKA_H_INCLUDED 27 28 #include "logthrdest/logthrdestdrv.h" 29 #include <librdkafka/rdkafka.h> 30 31 typedef struct 32 { 33 LogThreadedDestDriver super; 34 35 LogTemplateOptions template_options; 36 LogTemplate *key; 37 LogTemplate *message; 38 LogTemplate *topic_name; 39 GHashTable *topics; 40 GMutex topics_lock; 41 42 gboolean transaction_commit; 43 GList *config; 44 gchar *bootstrap_servers; 45 gchar *fallback_topic_name; 46 rd_kafka_topic_t *topic; 47 rd_kafka_t *kafka; 48 gint flush_timeout_on_shutdown; 49 gint flush_timeout_on_reload; 50 gint poll_timeout; 51 gboolean transaction_inited; 52 } KafkaDestDriver; 53 54 #define TOPIC_NAME_ERROR topic_name_error_quark() 55 56 GQuark topic_name_error_quark(void); 57 58 enum KafkaTopicError 59 { 60 TOPIC_LENGTH_ZERO, 61 TOPIC_DOT_TWO_DOTS, 62 TOPIC_EXCEEDS_MAX_LENGTH, 63 TOPIC_INVALID_PATTERN, 64 }; 65 66 void kafka_dd_set_topic(LogDriver *d, LogTemplate *topic); 67 gboolean kafka_dd_reopen(LogDriver *d); 68 void kafka_dd_set_fallback_topic(LogDriver *d, const gchar *fallback_topic); 69 void kafka_dd_merge_config(LogDriver *d, GList *props); 70 void kafka_dd_set_bootstrap_servers(LogDriver *d, const gchar *bootstrap_servers); 71 void kafka_dd_set_key_ref(LogDriver *d, LogTemplate *key); 72 void kafka_dd_set_message_ref(LogDriver *d, LogTemplate *message); 73 void kafka_dd_shutdown(LogThreadedDestDriver *s); 74 void kafka_dd_set_flush_timeout_on_shutdown(LogDriver *d, gint shutdown_timeout); 75 void kafka_dd_set_flush_timeout_on_reload(LogDriver *d, gint reload_timeout); 76 void kafka_dd_set_poll_timeout(LogDriver *d, gint poll_timeout); 77 void kafka_dd_set_transaction_commit(LogDriver *d, gboolean transaction_commit); 78 79 gboolean kafka_dd_validate_topic_name(const gchar *name, GError **error); 80 gboolean kafka_dd_is_topic_name_a_template(KafkaDestDriver *self); 81 rd_kafka_topic_t *kafka_dd_query_insert_topic(KafkaDestDriver *self, const gchar *name); 82 LogTemplateOptions *kafka_dd_get_template_options(LogDriver *d); 83 84 LogDriver *kafka_dd_new(GlobalConfig *cfg); 85 86 #endif 87