Packages

erlkaf

2.0.1
2.2.1 2.2.0 2.1.9 2.1.8 2.1.7 2.1.6 2.1.5 2.1.4 2.1.3 2.1.2 2.1.1 2.1.0 2.0.9 2.0.8 2.0.7 2.0.6 2.0.5 2.0.4 2.0.3 2.0.2 2.0.1 2.0.0 1.2.0 1.1.9 1.1.8 1.1.7 1.1.6 1.1.5 1.1.4 1.1.3

erlkaf - Erlang Kafka library based on librdkafka

HexDocs

Current section

Files

Jump to
Readme
30 Versions
3 Dependencies
Dependants
Files
Activity
erlkaf c_src queuecallbacksdispatcher.h
Raw

c_src/queuecallbacksdispatcher.h 

#ifndef C_SRC_QUEUE_CALLBACKS_DISPATCHER_H_

#define C_SRC_QUEUE_CALLBACKS_DISPATCHER_H_



#include "macros.h"

#include "critical_section.h"

#include <concurrentqueue/blockingconcurrentqueue.h>



#include <unordered_map>

#include <thread>



typedef struct rd_kafka_s rd_kafka_t;



class QueueCallbacksDispatcher

{

public:



    QueueCallbacksDispatcher();

    ~QueueCallbacksDispatcher();



    void watch(rd_kafka_t* instance, bool is_consumer);

    bool remove(rd_kafka_t* instance);

    void signal(rd_kafka_t* instance);



private:



    void process_callbacks();



    CriticalSection crt_;

    std::thread thread_callbacks_;



    bool running_;

    moodycamel::BlockingConcurrentQueue<rd_kafka_t*> events_;

    std::unordered_map<rd_kafka_t*, bool> objects_;



    DISALLOW_COPY_AND_ASSIGN(QueueCallbacksDispatcher);

};



#endif  // C_SRC_QUEUE_CALLBACKS_DISPATCHER_H_