增加支持推流的代码,并替换librtspclient.so

chenshijun

2019-07-23 5e85a7e80764e1a1eae39c8ce9bafed29d8773a9

增加支持推流的代码,并替换librtspclient.so

 apipassive.go

@@ -55,3 +55,20 @@

    return d, wid, hei
}

//GetAVPacket get AVPacket
func (h *GoFFMPEG) GetAVPacket() ([]byte, int, int) {
    var key C.int
    var size C.int

    p := C.wrap_fn_get_avpacket(h.ffmpeg, &size, &key)
    if size <= 0 {
        return nil, 0, -1
    }
    defer C.free(unsafe.Pointer(p))
    d := C.GoBytes(p, size)
    s := int(size)
    k := int(key)

    return d, s, k
}

 cffmpeg.h

@@ -26,6 +26,7 @@

void c_ffmpeg_build_decoder(const cffmpeg h);
void* c_ffmpeg_get_pic_decoder(const cffmpeg h, int *wid, int *hei);
void* c_ffmpeg_get_avpacket(const cffmpeg h, int *size, int *key);
////////////active api
void c_ffmpeg_active_recorder(const cffmpeg h, const char *dir, int mind, int maxd, rec_func fn);
void c_ffmpeg_active_decoder(const cffmpeg h, dec_func fn);

 csrc/cffmpeg.cpp

@@ -86,6 +86,13 @@
    s->GetPicDecoder(&data, wid, hei);
    return data;
}

void* c_ffmpeg_get_avpacket(const cffmpeg h, int *size, int *key){
    Wrapper *s = (Wrapper*)h;
    unsigned char *data = NULL;
    s->GetPacket(&data, size, key);
    return data;
}
/////////////////////active api
void c_ffmpeg_active_recorder(const cffmpeg h, const char *dir, int mind, int maxd, 
                                    rec_func fn){

 csrc/ffmpeg/format/FormatIn.cpp

@@ -21,7 +21,6 @@
#include "../data/FrameData.hpp"

#include "../../common/gpu/info.h"
#include "PsToEs.hpp"

using namespace logif;

@@ -90,9 +89,10 @@
    int FormatIn::openGb28181(const char *filename, AVDictionary **options){

        std::string fn = filename;
        addCamera(fn);
        //GB28181API gb28181(fn);
        handle_gb28181.addCamera(fn);

        int ret = openWithCustomIO(NULL, readData, options);
        int ret = openWithCustomIO((void *)&handle_gb28181, handle_gb28181.readData, options);
        if(ret < 0){
            logIt("do openWithCustomIO failed:%d",ret);
        }

 csrc/ffmpeg/format/FormatIn.hpp

@@ -3,6 +3,7 @@

#include <stdint.h>
#include <memory>
#include "PsToEs.hpp"

struct AVFormatContext;
struct AVDictionary;
@@ -62,6 +63,7 @@
        AVIOContext            *io_ctx_;
        uint8_t             *read_io_buff_;
        const int             read_io_buff_size_;
        GB28181API            handle_gb28181;
    };
}


 csrc/thirdparty/gb28181/include/PsToEs.hpp

@@ -67,7 +67,6 @@
        while (!q.empty()) q.pop_front();
        pthread_mutex_unlock(&mtx);
    }

private:
    deque<T> q;
    pthread_mutex_t mtx;
@@ -80,111 +79,115 @@
    int buffLen;
} frameBuffInfo;

MyQueue<frameBuffInfo *> m_rtpQueue;
long Handle;
class GB28181API{
public:
    GB28181API(/*string rtspUrl*/){
//        handle = addCamera(rtspUrl);
    }

bool pushInfo(unsigned char *data, int datalen) {
    ~GB28181API(){
        printf("GB28181API end!\n");
        m_rtpQueue.clearAll();
        deleteCamera();
    }

    frameBuffInfo *info = new frameBuffInfo();
    info->buff = new unsigned char[datalen];
    info->buffLen = datalen;
    memcpy(info->buff, data, datalen);

    //printf(" m_rtpQueue.push befores ");
    m_rtpQueue.push(info);
    //printf(" m_rtpQueue.push after ");

    return true;
}

int readData(void *opaque, unsigned char *buf, int bufsize) {

//    GB28181API *_this = (GB28181API *)opaque;
    int len = bufsize;
    int diff = 0;
    do {

    //printf(" m_rtpQueue.pop before ");
    //从缓存中获取buffinfo
    frameBuffInfo *buffinfo = m_rtpQueue.pop();
//        DBG(" m_rtpQueue.pop after ");
    diff = len - buffinfo->buffLen;

    //帧长大于bufsize
    if (diff < 0) {
        printf("/帧长大于bufsize:%d\n", diff);
        memcpy(buf + bufsize - len, buffinfo->buff, len);
    bool pushInfo(unsigned char *data, int datalen) {

        frameBuffInfo *info = new frameBuffInfo();
        info->buffLen = buffinfo->buffLen - len;
        info->buff = new unsigned char[buffinfo->buffLen - len]{};
        memcpy(info->buff, buffinfo->buff + len, buffinfo->buffLen - len);
        info->buff = new unsigned char[datalen];
        info->buffLen = datalen;
        memcpy(info->buff, data, datalen);

        m_rtpQueue.push_front_one(info);
    } else if (diff == 0) {
        printf("/帧长等于bufsize:%d\n", diff);
        memcpy(buf + bufsize - len, buffinfo->buff, buffinfo->buffLen);
    } else if (diff > 0) {
        printf("/帧长小于bufsize:%d\n", diff);
        memcpy(buf + bufsize - len, buffinfo->buff, buffinfo->buffLen);
        len = len - buffinfo->buffLen;   //还需要填充的大小
        memset(buf + bufsize - len, 0, len);
        //不等待填充，直接进行解码
        diff = 0;
    }
    delete[] buffinfo->buff;
    delete buffinfo;
    } while (diff > 0);
        //printf(" m_rtpQueue.push befores ");
        m_rtpQueue.push(info);
        //printf(" m_rtpQueue.push after ");

    return bufsize;
}

void streamCallBack(int datatype, int frametype, unsigned char *data, unsigned int datalen, long userdata)
{
    //GB28181API *_this = (GB28181API *)userdata;
    printf("userdata:%ld,datatype:%d, frametype:%d,  datalen:%d\n", userdata, datatype, frametype, datalen);

//    //debug===============
//    static int count = 0;
//    static FILE *fp_write = NULL;
//    if(count < 100) {
//        count++;
//
//        if (!fp_write) {
//            fp_write = fopen("stream_callback.mp4", "wb+");
//        }
//
//        fwrite(data, sizeof(char), datalen, fp_write);
//    }
//    if(count >= 100){
//        if (!fp_write) {
//            fclose(fp_write);
//        }
//    }
//    //debug===============

    static bool startFlag = false;
    if(frametype == GB_VIDEO_FRAME_I){
        startFlag = true;
    }
    if((data != NULL) && (startFlag == true)){
        pushInfo(data, datalen);
        return true;
    }

}
    static int readData(void *opaque, unsigned char *buf, int bufsize) {

long addCamera(string &rtsp){
    printf("RTSPSTREAM_Open\n");
    long userdata = 1001;
    Handle = RTSPSTREAM_Open(rtsp.c_str(), streamCallBack, userdata);
    return Handle;
}
        GB28181API *_this = (GB28181API *)opaque;
        int len = bufsize;
        int diff = 0;
        do {
//            printf(" m_rtpQueue.pop before \n");
            //从缓存中获取buffinfo
            frameBuffInfo *buffinfo = _this->m_rtpQueue.pop();
//            printf(" m_rtpQueue.pop after \n");
            diff = len - buffinfo->buffLen;

void deleteCamera(void){
    m_rtpQueue.clearAll();
    RTSPSTREAM_Close(Handle);
    Handle = 0;
}
            //帧长大于bufsize
            if (diff < 0) {
//                printf("/帧长大于bufsize:%d\n", diff);
                memcpy(buf + bufsize - len, buffinfo->buff, len);

                frameBuffInfo *info = new frameBuffInfo();
                info->buffLen = buffinfo->buffLen - len;
                info->buff = new unsigned char[buffinfo->buffLen - len]{};
                memcpy(info->buff, buffinfo->buff + len, buffinfo->buffLen - len);

//                printf("/帧长大于info->buffLen:%d\n", info->buffLen);
                _this->m_rtpQueue.push_front_one(info);
//                printf("/帧长大于info->buffLen\n");
            } else if (diff == 0) {
//                printf("/帧长等于bufsize:%d\n", diff);
                memcpy(buf + bufsize - len, buffinfo->buff, buffinfo->buffLen);
            } else if (diff > 0) {
//                printf("/帧长小于bufsize:%d\n", diff);
                memcpy(buf + bufsize - len, buffinfo->buff, buffinfo->buffLen);
                len = len - buffinfo->buffLen;   //还需要填充的大小
                memset(buf + bufsize - len, 0, len);
                //不等待填充，直接进行解码
                diff = 0;
            }
            delete[] buffinfo->buff;
            delete buffinfo;
//            printf("/帧长大于info->buffLen1\n");
        } while (diff > 0);

        return bufsize;
    }

    static void streamCallBack(int datatype, int frametype, unsigned char *data, unsigned int datalen, long userdata)
    {
        GB28181API *_this = (GB28181API *)userdata;
        //printf("userdata:%ld,datatype:%d, frametype:%d,  datalen:%d\n", userdata, datatype, frametype, datalen);

        /*static FILE* fp_write = NULL;
        if (!fp_write)

        {

            fp_write = fopen("stream_callback.mp4", "wb+");

        }

        fwrite(data, sizeof(char), datalen, fp_write);    */

        static bool startFlag = false;
        if(frametype == GB_VIDEO_FRAME_I){
            startFlag = true;
        }
        if((data != NULL) && (startFlag == true)){
                _this->pushInfo(data, datalen);
        }
    }

    void addCamera(string &rtsp){
//        long userdata = 1001;//
        printf("RTSPSTREAM_Open\n");
        handle = RTSPSTREAM_Open(rtsp.c_str(), streamCallBack, (long)this);
//        return handle;
    }

    void deleteCamera(){
        printf("RTSPSTREAM_Close\n");
        RTSPSTREAM_Close(handle);
    }	
private:
    MyQueue<frameBuffInfo *> m_rtpQueue;
    long handle;
};



 csrc/thirdparty/gb28181/lib/librtspclient.so

Binary files differ

 csrc/wrapper.cpp

@@ -218,6 +218,9 @@
                    pkt.id = id++;
                }
                pkt.data = data;
                if(data != nullptr) {
                    cacheAVPacket(data->getAVPacket());
                }

                run_worker(in.get(), pkt);
                if(!data){
@@ -334,6 +337,30 @@
        auto p = list_pic_.front();
        *data = p.data; *w = p.w; *h = p.h;
        list_pic_.pop_front();
    }

    void Wrapper::GetPacket(unsigned char **pktData, int *size, int *key){
        std::lock_guard<std::mutex> l(mutex_avpkt_);
        if(list_avpkt_.empty()){
            return;
        }
        auto pkt = list_avpkt_.front();
        *key = pkt.flags & AV_PKT_FLAG_KEY;
        *size = pkt.size;
        *pktData = (unsigned char *)malloc(*size);
        memcpy(*pktData, pkt.data, pkt.size);

        list_avpkt_.pop_front();
    }
    void Wrapper::cacheAVPacket(const AVPacket &pkt){
        std::lock_guard<std::mutex> l(mutex_pic_);
        while(list_avpkt_.size() > 10){
//            printf("cacheAVPacket drop packets!!!!!!!!!!\n");
            for(int i = 0; i < 5; i++){
                list_avpkt_.pop_front();
            }
        }
        list_avpkt_.emplace_back(pkt);
    }

    void Wrapper::run_worker(ffwrapper::FormatIn *in, avpacket &pkt){
@@ -523,7 +550,7 @@
            }
            *size = pkt.size + extradata_size;
            *out = (unsigned char *)malloc(*size);
	

            memcpy(*out, extra, extradata_size);
            memcpy(*out + extradata_size, pkt.data, pkt.size);


 csrc/wrapper.hpp

@@ -1,6 +1,10 @@
#ifndef _cffmpeg_wrapper_hpp_
#define _cffmpeg_wrapper_hpp_

extern "C"{
#include <libavcodec/avcodec.h>
}

#include <stdint.h>

#include <string>
@@ -58,6 +62,8 @@

            void cache_rec_info(int &index, std::string &path);
            void cache_pic(std::shared_ptr<ffwrapper::FrameData> &frame);

            void cacheAVPacket(const AVPacket &pkt);
        public: 
            int RunStream(const char* input);
        private: 
@@ -78,6 +84,7 @@
        public: //decoder
            void BuildDecoder();
            void GetPicDecoder(unsigned char **data, int *w, int *h);
            void GetPacket(unsigned char **pktData, int *size, int *key);
            //active api
            void ActiveDecoder(FUNC_DEC fn);

@@ -106,6 +113,10 @@

            std::list<pic_bgr24> list_pic_;
            std::mutex mutex_pic_;

            std::list<AVPacket> list_avpkt_;
            std::mutex mutex_avpkt_;

            // active api
            FUNC_REC func_rec_;
            FUNC_DEC func_dec_;

 libcffmpeg.c

@@ -39,6 +39,8 @@
        release_if_err(fn_decoder, lib);
        fn_decoder_pic = (lib_cffmpeg_pic)dlsym(lib, "c_ffmpeg_get_pic_decoder");
        release_if_err(fn_decoder_pic, lib);
        fn_get_avpacket = (lib_cffmpeg_avpacket)dlsym(lib, "c_ffmpeg_get_avpacket");
        release_if_err(fn_get_avpacket, lib);
        fn_active_recorder = (lib_cffmpeg_active_recorder)dlsym(lib, "c_ffmpeg_active_recorder");
        release_if_err(fn_active_recorder, lib);
        fn_active_decoder = (lib_cffmpeg_active_decoder)dlsym(lib, "c_ffmpeg_active_decoder");
@@ -109,6 +111,10 @@
    return fn_decoder_pic(h, wid, hei);
}

void* wrap_fn_get_avpacket(const cffmpeg h, int* size, int* key){
    return fn_get_avpacket(h, size, key);
}

void wrap_fn_active_recorder(const cffmpeg h, const char* dir, int mind, int maxd, rec_func fn){
    fn_active_recorder(h, dir, mind, maxd, fn);
}

 libcffmpeg.h

@@ -24,6 +24,7 @@
typedef char*(*lib_cffmpeg_info_recorder)(const cffmpeg, int*, int*);
typedef void (*lib_cffmpeg_decoder)(const cffmpeg);
typedef void*(*lib_cffmpeg_pic)(const cffmpeg, int*, int*);
typedef void*(*lib_cffmpeg_avpacket)(const cffmpeg, int*, int*);
typedef void (*lib_cffmpeg_active_recorder)(const cffmpeg, const char*, int, int, rec_func);
typedef void (*lib_cffmpeg_active_decoder)(const cffmpeg, dec_func);
typedef void*(*lib_cffmpeg_decode_jpeg)(const cffmpeg, const char*, int*, int*);
@@ -39,6 +40,7 @@
static lib_cffmpeg_info_recorder       fn_info_recorder = NULL;
static lib_cffmpeg_decoder             fn_decoder = NULL;
static lib_cffmpeg_pic                 fn_decoder_pic = NULL;
static lib_cffmpeg_avpacket            fn_get_avpacket = NULL;
static lib_cffmpeg_active_recorder     fn_active_recorder = NULL;
static lib_cffmpeg_active_decoder      fn_active_decoder = NULL;
static lib_cffmpeg_decode_jpeg         fn_dec_jpeg = NULL;
@@ -58,6 +60,7 @@
char* wrap_fn_info_recorder(const cffmpeg, int*, int*);
void wrap_fn_decoder(const cffmpeg h);
void* wrap_fn_decoder_pic(const cffmpeg h, int* wid, int* hei);
void* wrap_fn_get_avpacket(const cffmpeg h, int* size, int* key);
void wrap_fn_active_recorder(const cffmpeg h, const char* dir, int mind, int maxd, rec_func fn);
void wrap_fn_active_decoder(const cffmpeg h, dec_func fn);
void* wrap_fn_decode_jpeg(const cffmpeg h, const char* file, int* wid, int* hei);