diff --git a/trunk/src/app/srs_app_kafka.cpp b/trunk/src/app/srs_app_kafka.cpp index 7dffec4be..f94f1ef88 100644 --- a/trunk/src/app/srs_app_kafka.cpp +++ b/trunk/src/app/srs_app_kafka.cpp @@ -28,6 +28,8 @@ #include #include +#ifdef SRS_AUTO_KAFKA + SrsKafkaProducer::SrsKafkaProducer() { worker = new SrsAsyncCallWorker(); @@ -66,3 +68,5 @@ void SrsKafkaProducer::stop() worker->stop(); } +#endif + diff --git a/trunk/src/app/srs_app_kafka.hpp b/trunk/src/app/srs_app_kafka.hpp index 0bbd7b570..1fccf879a 100644 --- a/trunk/src/app/srs_app_kafka.hpp +++ b/trunk/src/app/srs_app_kafka.hpp @@ -31,6 +31,11 @@ class SrsAsyncCallWorker; +#ifdef SRS_AUTO_KAFKA + +/** + * the kafka producer used to save log to kafka cluster. + */ class SrsKafkaProducer { private: @@ -45,3 +50,6 @@ public: }; #endif + +#endif + diff --git a/trunk/src/app/srs_app_server.cpp b/trunk/src/app/srs_app_server.cpp index d6c315a24..c88884551 100755 --- a/trunk/src/app/srs_app_server.cpp +++ b/trunk/src/app/srs_app_server.cpp @@ -509,7 +509,9 @@ SrsServer::SrsServer() #ifdef SRS_AUTO_INGEST ingester = NULL; #endif +#ifdef SRS_AUTO_KAFKA kafka = new SrsKafkaProducer(); +#endif } SrsServer::~SrsServer() @@ -539,7 +541,9 @@ void SrsServer::destroy() srs_freep(ingester); #endif +#ifdef SRS_AUTO_KAFKA srs_freep(kafka); +#endif if (pid_fd > 0) { ::close(pid_fd); @@ -565,7 +569,9 @@ void SrsServer::dispose() ingester->dispose(); #endif +#ifdef SRS_AUTO_KAFKA kafka->stop(); +#endif SrsSource::dispose_all(); @@ -874,12 +880,19 @@ int SrsServer::start_kafka() { int ret = ERROR_SUCCESS; +#ifdef SRS_AUTO_KAFKA if ((ret = kafka->initialize()) != ERROR_SUCCESS) { srs_error("initialize the kafka producer failed. ret=%d", ret); return ret; } - return kafka->start(); + if ((ret = kafka->start()) != ERROR_SUCCESS) { + srs_error("start kafka failed. ret=%d", ret); + return ret; + } +#endif + + return ret; } int SrsServer::cycle() diff --git a/trunk/src/app/srs_app_server.hpp b/trunk/src/app/srs_app_server.hpp index a2f131154..04cf9644a 100644 --- a/trunk/src/app/srs_app_server.hpp +++ b/trunk/src/app/srs_app_server.hpp @@ -55,7 +55,9 @@ class SrsTcpListener; #ifdef SRS_AUTO_STREAM_CASTER class SrsAppCasterFlv; #endif +#ifdef SRS_AUTO_KAFKA class SrsKafkaProducer; +#endif // listener type for server to identify the connection, // that is, use different type to process the connection. @@ -248,7 +250,9 @@ private: #ifdef SRS_AUTO_INGEST SrsIngester* ingester; #endif +#ifdef SRS_AUTO_KAFKA SrsKafkaProducer* kafka; +#endif private: /** * the pid file fd, lock the file write when server is running. diff --git a/trunk/src/protocol/srs_kafka_stack.cpp b/trunk/src/protocol/srs_kafka_stack.cpp index e6fdc9b97..f23d0d00a 100644 --- a/trunk/src/protocol/srs_kafka_stack.cpp +++ b/trunk/src/protocol/srs_kafka_stack.cpp @@ -25,6 +25,8 @@ using namespace std; +#ifdef SRS_AUTO_KAFKA + SrsKafkaString::SrsKafkaString() { size = -1; @@ -194,3 +196,5 @@ SrsKafkaTopicMetadataRequest::~SrsKafkaTopicMetadataRequest() { } +#endif + diff --git a/trunk/src/protocol/srs_kafka_stack.hpp b/trunk/src/protocol/srs_kafka_stack.hpp index e8574161f..e5df62b68 100644 --- a/trunk/src/protocol/srs_kafka_stack.hpp +++ b/trunk/src/protocol/srs_kafka_stack.hpp @@ -31,6 +31,8 @@ #include +#ifdef SRS_AUTO_KAFKA + // https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ApiKeys enum SrsKafkaApiKey { @@ -310,3 +312,5 @@ public: #endif +#endif +