fix rec

zhangmeng

2019-08-19 54ea1c13885725584a6a50d520f67e8a75f85b6f

 csrc/wrapper.cpp

@@ -149,7 +149,10 @@
                auto data(std::make_shared<CodedData>());
               if(!in->readPacket(data)){
                    logIt("read packet error");
                    pkt.id = -1; data = nullptr; id = 0;
                    data.reset();
                    data = nullptr;
                    pkt.id = -1; 
                    id = 0;
               }else{
                    pkt.id = id++;
                }
@@ -160,7 +163,10 @@

                run_worker(in.get(), pkt);
                if(!data){
                    map_rec_.clear();
                    {
                        std::lock_guard<std::mutex> l(mutex_rec_);
                        map_rec_.clear();
                    }
                    std::lock_guard<std::mutex> locker(mtx_rec_pkt_);
                    list_rec_pkt_.clear();

@@ -205,20 +211,25 @@
            }
        }
        cache_rec_pkt(pkt);
        for(auto &i : map_rec_){
            if (!i.second.rec){
                i.second.rec = i.second.fn_init(in);
                if (i.second.rec){
                    std::lock_guard<std::mutex> locker(mtx_rec_pkt_);
                    for(auto &k : list_rec_pkt_){
                        avpacket p = {k.data, k.id};
                        i.second.rec->CachePacket(p);
        {
            std::lock_guard<std::mutex> l(mutex_rec_);

            for(auto &i : map_rec_){
                if (!i.second.rec){
                    i.second.rec = std::move(init_recorder(in, i.second.rid, i.second.dir, i.second.min, i.second.max));
                    if (i.second.rec){
                        std::lock_guard<std::mutex> locker(mtx_rec_pkt_);
                        for(auto &k : list_rec_pkt_){
                            avpacket p = {k.data, k.id};
                            i.second.rec->CachePacket(p);
                        }
                        logIt("START REC %d FRAMES", list_rec_pkt_.size());
                    }
                    logIt("START REC %d FRAMES", list_rec_pkt_.size());
                }else if (i.second.rec){
                    i.second.rec->CachePacket(pkt);
                }
            }else if (i.second.rec){
                i.second.rec->CachePacket(pkt);
            }

        }
    }

@@ -255,13 +266,13 @@
    }

    //////////////recorder
    std::shared_ptr<Recorder> Wrapper::init_recorder(FormatIn *in, std::string id, std::string dir, const int mind, const int maxd){
    std::unique_ptr<Recorder> Wrapper::init_recorder(FormatIn *in, std::string id, std::string dir, const int mind, const int maxd){
        if(!in){
            logIt("Init wrapper first");
            return nullptr;
        }

        auto rec = std::make_shared<Recorder>(in, id.c_str());
        std::unique_ptr<Recorder> rec(new Recorder(in, id.c_str()));

        rec->SetCallback([&](std::string &id, int &index, std::string &path){
            cache_rec_info(id, index, path);
@@ -283,25 +294,50 @@
        std::string rid(id);
        std::string dir(output);
        
        auto fn = [=](FormatIn *in){
            return init_recorder(in, rid, dir, mindur, maxdur);
        };
        std::shared_ptr<Recorder> rec(nullptr);
        
        FnRec r = FnRec{fn, rec};
        map_rec_[rid] = r;
        std::lock_guard<std::mutex> l(mutex_rec_);

        // auto fn = [=](FormatIn *in){
        //     return init_recorder(in, rid, dir, mindur, maxdur);
        // };
        // FnRec r = FnRec{fn, nullptr};
        if (map_rec_.find(rid) != map_rec_.end()){
            map_rec_.erase(rid);
        }
        // for (auto iter = map_rec_.begin(); iter != map_rec_.end();){
        //     if (iter->second.rec && iter->second.rec->ErrorOcurred()){
        //         iter == map_rec_.erase(iter);
        //     }else{
        //         iter++;
        //     }
        // }
        FnRec fr;
        fr.rid = rid;
        fr.dir = dir;
        fr.min = mindur;
        fr.max = maxdur;
        map_rec_[rid] = std::move(fr);

        minduration = mindur * 25;
        maxduration = maxdur * 25;
    }

    int Wrapper::FireRecorder(const char* sid,const int64_t &id){
        std::lock_guard<std::mutex> l(mutex_rec_);

        auto iter = map_rec_.find(sid);
        if (iter != map_rec_.end()){
            if(iter->second.rec){
                iter->second.rec->FireRecorder(id);
            }
        }

        // for (auto iter = map_rec_.begin(); iter != map_rec_.end();){
        //     if (iter->second.rec && iter->second.rec->ErrorOcurred()){
        //         iter == map_rec_.erase(iter);
        //     }else{
        //         iter++;
        //     }
        // }
    }

    void Wrapper::cache_rec_info(std::string &id, int &index, std::string &path){
@@ -336,10 +372,21 @@

        if (map_rec_.find(info.rec_id) != map_rec_.end())
            map_rec_.erase(info.rec_id);

        for (auto iter = map_rec_.begin(); iter != map_rec_.end();){
            if (iter->second.rec && iter->second.rec->ErrorOcurred()){
                iter == map_rec_.erase(iter);
            }else{
                iter++;
            }
        }

        // logIt("go get info index: %d, file: %s\n", index, path.c_str());
    }

    std::string Wrapper::GetRecorderID(const std::string &path){
        std::lock_guard<std::mutex> l(mutex_rec_);

        std::string ret("");
        auto iter = list_rec_map_.find(path);
        if (iter != list_rec_map_.end()){