external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

refine kafka, simplify code.

Browse source
This commit is contained in:
winlin 2015-10-23 14:30:16 +08:00
parent 9a47390253
commit 71451878c9
2 changed files with 23 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

40
trunk/src/app/srs_app_kafka.cpp
View file

@ -178,41 +178,31 @@ int SrsKafkaPartition::flush(SrsKafkaPartitionCache* pc)
return kafka->write_messages(topic, id, *pc); return kafka->write_messages(topic, id, *pc);
} }
SrsKafkaMessage::SrsKafkaMessage(int k) SrsKafkaMessage::SrsKafkaMessage(SrsKafkaProducer* p, int k, SrsJsonObject* j)
{ {
producer = p;
key = k; key = k;
obj = j;
} }
SrsKafkaMessage::~SrsKafkaMessage() SrsKafkaMessage::~SrsKafkaMessage()
{ {
srs_freep(obj);
} }
SrsKafkaMessageOnClient::SrsKafkaMessageOnClient(SrsKafkaProducer* p, int k, SrsListenerType t, string i) int SrsKafkaMessage::call()
: SrsKafkaMessage(k)
{ {
producer = p; int ret = producer->send(key, obj);
type = t;
ip = i;
}
SrsKafkaMessageOnClient::~SrsKafkaMessageOnClient()
{
}
int SrsKafkaMessageOnClient::call()
{
SrsJsonObject* obj = SrsJsonAny::object();
obj->set("msg", SrsJsonAny::str("accept")); // the obj is manged by producer now.
obj->set("type", SrsJsonAny::integer(type)); obj = NULL;
obj->set("ip", SrsJsonAny::str(ip.c_str()));
return producer->send(key, obj); return ret;
} }
string SrsKafkaMessageOnClient::to_string() string SrsKafkaMessage::to_string()
{ {
return ip; return "kafka";
} }
SrsKafkaCache::SrsKafkaCache() SrsKafkaCache::SrsKafkaCache()
@ -393,7 +383,13 @@ void SrsKafkaProducer::stop()
int SrsKafkaProducer::on_client(int key, SrsListenerType type, string ip) int SrsKafkaProducer::on_client(int key, SrsListenerType type, string ip)
{ {
return worker->execute(new SrsKafkaMessageOnClient(this, key, type, ip)); SrsJsonObject* obj = SrsJsonAny::object();
obj->set("msg", SrsJsonAny::str("accept"));
obj->set("type", SrsJsonAny::integer(type));
obj->set("ip", SrsJsonAny::str(ip.c_str()));
return worker->execute(new SrsKafkaMessage(this, key, obj));
} }
int SrsKafkaProducer::send(int key, SrsJsonObject* obj) int SrsKafkaProducer::send(int key, SrsJsonObject* obj)

18
trunk/src/app/srs_app_kafka.hpp
View file

@ -80,21 +80,13 @@ public:
*/ */
class SrsKafkaMessage : public ISrsAsyncCallTask class SrsKafkaMessage : public ISrsAsyncCallTask
{ {
protected: private:
int key;
public:
SrsKafkaMessage(int k);
virtual ~SrsKafkaMessage();
};
struct SrsKafkaMessageOnClient : public SrsKafkaMessage
{
public:
SrsKafkaProducer* producer; SrsKafkaProducer* producer;
SrsListenerType type; int key;
std::string ip; SrsJsonObject* obj;
public: public:
SrsKafkaMessageOnClient(SrsKafkaProducer* p, int k, SrsListenerType t, std::string i); SrsKafkaMessage(SrsKafkaProducer* p, int k, SrsJsonObject* j);
virtual ~SrsKafkaMessageOnClient(); virtual ~SrsKafkaMessage();
// interface ISrsAsyncCallTask // interface ISrsAsyncCallTask
public: public:
virtual int call(); virtual int call();