Configurable compression (voiceip#11)

Author: kingster

README.md

@@ -11,6 +11,7 @@ Install this plugin to publish all of the freeswitch generated events to Kafka r
 		<param name="bootstrap-servers" value="localhost:9092"/>
 		<param name="topic-prefix" value="topic_prefix"/>
 		<param name="buffer-size" value="256" /> 
+		<param name="compression" value="snappy"/>
 	</settings>
  </configuration>
 ```

event_kafka.conf.xml

@@ -6,5 +6,6 @@
 		<param name="username" value="" /> <!-- set it only if you have sasl enabled on your kafka cluster -->
 		<param name="password" value="" />
 		<param name="buffer-size" value="16" /> 
+		<param name="compression" value="snappy"/>
 	</settings>
  </configuration>

mod_event_kafka.cpp

@@ -57,6 +57,8 @@ namespace mod_event_kafka {
                             "fs", NULL, "topic-prefix", "Kafka Topic Prefix"),
         SWITCH_CONFIG_ITEM("buffer-size", SWITCH_CONFIG_INT, CONFIG_RELOADABLE, &globals.buffer_size,
                             10, NULL, "buffer-size", "queue.buffering.max.messages"),
+        SWITCH_CONFIG_ITEM("compression", SWITCH_CONFIG_STRING, CONFIG_RELOADABLE, &globals.compression,
+                            "snappy", NULL, "snappy / lz4 ", "Compression"),
         SWITCH_CONFIG_ITEM_END()
     };
 
@@ -67,7 +69,8 @@ namespace mod_event_kafka {
             switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Could not open event_kafka.conf\n");
             return SWITCH_STATUS_FALSE;
         } else {
-            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_kafka.conf loaded [brokers: %s, topic/topic_prefix: %s/%s, username: %s, buffer-size: %d]", globals.brokers, globals.topic, globals.topic_prefix, globals.username, globals.buffer_size);
+            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "event_kafka.conf loaded [brokers: %s, topic/topic_prefix: %s/%s, username: %s, buffer-size: %d, compression: %s]", 
+            globals.brokers, globals.topic, globals.topic_prefix, globals.username, globals.buffer_size, globals.compression);
         }
         return SWITCH_STATUS_SUCCESS;
     }
@@ -97,7 +100,7 @@ namespace mod_event_kafka {
                 switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, errstr);
             }
 
-            if (rd_kafka_conf_set(conf, "compression.codec", "snappy", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+            if (rd_kafka_conf_set(conf, "compression.codec", globals.compression, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
                 switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, errstr);
             }
 

mod_event_kafka.hpp

@@ -14,6 +14,7 @@ namespace mod_event_kafka {
 		char *username;
 		char *password;
 		int buffer_size;
+		char *compression;
 	} globals;
 
 	SWITCH_MODULE_LOAD_FUNCTION(mod_event_kafka_load);