Class Consumer::ConsumerImpl

phxqueue/consumer/consumer.cpp:56–86

};

class Consumer::ConsumerImpl {
  public:
    ConsumerImpl() {}
    virtual ~ConsumerImpl() {}

    std::string topic = "";
    int topic_id = -1;
    comm::FactoryList<comm::HandlerFactory> fs;
    HeartBeatLock lock;
    FreqMan freq;
    ConsumerOption opt;

    int vpid = -1;
    comm::proto::ConsumerContext cc;

    int consume_fds[2];

    vector<shared_ptr<comm::proto::QItem> > items;
    vector<comm::HandleResult> handle_results;

    DispatchCtx_t dispatch_ctx;
    unique_ptr<ConsumeCtx_t[]> handle_ctxs;
    unique_ptr<ConsumeCtx_t[]> batch_handle_ctxs;
    unique_ptr<queue<int>[]> handle_buckets;
    unique_ptr<bool[]> batch_handle_finish;

	stCoCond_t *cond = nullptr;
    int nhandle_task_finished = 0;
    int nbatch_handle_task_finished = 0;

};

Consumer::Consumer(const ConsumerOption &opt) : impl_(new ConsumerImpl()) {
    assert(impl_);