update

zhangmeng

2019-11-12 c5a01eed95f1837e93fee27bce4da78c79f4ed10

 csrc/thirdparty/gb28181/include/PsToEs.hpp

@@ -5,6 +5,7 @@
#include <pthread.h>

#include "librtsp.h"
#include <sys/time.h>


using namespace std;
@@ -14,7 +15,7 @@
public:
    MyQueue():mtx(PTHREAD_MUTEX_INITIALIZER), cond(PTHREAD_COND_INITIALIZER){
        t.tv_sec = 0;
        t.tv_nsec = 20000000;
        t.tv_nsec = 0;
    }

    ~MyQueue() {
@@ -39,9 +40,16 @@

    T pop() {
        pthread_mutex_lock(&mtx);
        while (q.empty()) {
            pthread_cond_wait(&cond, &mtx);
        if (q.empty()) {
            gettimeofday(&now, NULL);
            t.tv_sec = now.tv_sec + 3;
            t.tv_nsec = now.tv_usec * 1000;
//            pthread_cond_wait(&cond, &mtx);
            pthread_cond_timedwait(&cond, &mtx, &t);
        }
        if (q.empty()) {
            pthread_mutex_unlock(&mtx);
            return 0;
        }
        T value = q.front();
        q.pop_front();
@@ -67,12 +75,22 @@
      while (!q.empty()) q.pop_front();
      pthread_mutex_unlock(&mtx);
    }

	
   void clearAll(std::function<void(T)> fn){
      pthread_mutex_lock(&mtx);
      while (!q.empty()){
         T value = q.front();
         fn(value);
         q.pop_front();
      } 
      pthread_mutex_unlock(&mtx);
   }
private:
    deque<T> q;
    pthread_mutex_t mtx;
    pthread_cond_t cond;
    timespec t;
    struct timeval now;
};

typedef struct _buffInfo {
@@ -80,111 +98,127 @@
   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();
      m_rtpQueue.clearAll([](frameBuffInfo *info){
         delete[] info->buff;
         delete info;
      });
      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");
            if(buffinfo != nullptr){
         diff = len - buffinfo->buffLen;
            }else{
                return 0;
            }

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);
      }
   }

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

   void deleteCamera(){
      printf("RTSPSTREAM_Close\n");
      if(handle != -1){
         RTSPSTREAM_Close(handle);
      }

      handle = -1;
   }	
private:
   MyQueue<frameBuffInfo *> m_rtpQueue;
   long handle = -1;
};