ringbuffer

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <unistd.h>
using namespace std;
#define TAG "zj"

#define LOGV(...) __android_log_print(ANDROID_LOG_VERBOSE, TAG, __VA_ARGS__)
#define LOGD(...) __android_log_print(ANDROID_LOG_DEBUG, TAG, __VA_ARGS__)
#define LOGI(...) __android_log_print(ANDROID_LOG_INFO, TAG, __VA_ARGS__)
#define LOGW(...) __android_log_print(ANDROID_LOG_WARN, TAG, __VA_ARGS__)
#define LOGE(...) __android_log_print(ANDROID_LOG_ERROR, TAG, __VA_ARGS__)

//过于简单了
// readindex he writeindex

class CGBlockRingBuffer
{
public:
    CGBlockRingBuffer()
        :buffer(nullptr)
        , bufferSize(0)
        , writeIndex(0)
        , readIndex(0)
        , bytesCanRead(0)
    {

    }

    void Init(int buffSize)
    {
        LOGE("block init--%d-->",buffSize);
        buffer = new uint8_t[buffSize];
        if (!buffer) return;
        this->bufferSize = buffSize;
    }



    void Write(uint8_t* src, int size)
    {
        if (!src || size == 0) return;
        std::lock_guard<std::mutex> lk(lock);
        if (size >= bufferSize - bytesCanRead) {
            usleep(50);
            LOGE("buffer now is full  %d,%d,%d!",size ,bufferSize , bytesCanRead);
            return;
        }

        int bytesCanWrite = size < bufferSize - bytesCanRead ? size : (bufferSize - bytesCanRead);

        //if(bytesCanWrite)  如果可写的小于--- 则丢弃--
        //这个
        if (bytesCanWrite <= bufferSize - writeIndex)
        {
            memcpy(buffer + writeIndex, src, bytesCanWrite);
            writeIndex += bytesCanWrite;
            if (writeIndex == bufferSize)
            {
                writeIndex = 0;
            }
        }
        else
        {
            int room = bufferSize - writeIndex;
            memcpy(buffer + writeIndex, src, room);
            int left = bytesCanWrite - room;
            memcpy(buffer, src + room, left);
            writeIndex = left;
        }
        bytesCanRead += bytesCanWrite;
    }
    //先这样--to thi
    // 如果没有找到的话sleep一会儿---,to thi
    //1、先直接换ring 2、加这个优化 
    int getNotRead(uint8_t* dst, int size) {
        LOGE("getNotRead %d\n",size);
        LOGE("dst %d\n",dst);
        if (!dst || size == 0) {
            LOGE("return-----");
            return 0;
        }
        std::lock_guard<std::mutex> lk(lock);


        int bytesRead = size < bytesCanRead ? size : bytesCanRead;
        if (bytesRead <= bufferSize - readIndex)
        {
            memcpy(dst, buffer + readIndex, bytesRead);
            //readIndex += bytesRead;
            //if (readIndex == bufferSize) readIndex = 0;
        }
        else
        {
            int bytesHead = bufferSize - readIndex;
            memcpy(dst, buffer + readIndex, bytesHead);
            int bytesTail = bytesRead - bytesHead;
            memcpy(dst + readIndex, buffer, bytesTail);
            //readIndex = bytesTail;
        }
        //bytesCanRead -= bytesRead;
        return bytesRead;
    }

    int Read(uint8_t* dst, int size)
    {
        //read小于则等待
        //就看这两个的--
         LOGE("Read %d\n",size);
        if (!dst || size == 0) return 0;
        std::lock_guard<std::mutex> lk(lock);

        if(size > bytesCanRead ){
            usleep(50);
            LOGE("Buffer is empty!---->");
        }
        int bytesRead = size < bytesCanRead ? size : bytesCanRead;
        if (bytesRead <= bufferSize - readIndex)
        {
            memcpy(dst, buffer + readIndex, bytesRead);
            readIndex += bytesRead;
            if (readIndex == bufferSize) readIndex = 0;
        }
        else
        {
            int bytesHead = bufferSize - readIndex;
            memcpy(dst, buffer + readIndex, bytesHead);
            int bytesTail = bytesRead - bytesHead;
            memcpy(dst + readIndex, buffer, bytesTail);
            readIndex = bytesTail;
        }
        bytesCanRead -= bytesRead;
        return bytesRead;
    }
private:
    uint8_t* buffer;
    int bufferSize;
    int readIndex;
    int  writeIndex;
    int bytesCanRead;
    std::mutex lock;
};

//有了这个声音也没啥问题了

#ifndef _CRT_SECURE_NO_WARNINGS
#define _CRT_SECURE_NO_WARNINGS
#endif
#include <android/log.h>
#include "xop/RtmpServer.h"
#include "xop/HttpFlvServer.h"
#include "xop/RtmpPublisher.h"
#include "xop/RtmpClient.h"
#include "xop/HttpFlvServer.h"
#include "xop/H264Parser.h"
#include "net/EventLoop.h"
#include <errno.h>

#include "CGBlockRingBuffer.h"

#define TEST_RTMP_PUSHER  1
#define TEST_RTMP_CLIENT  0
#define TEST_MULTI_THREAD 0
#define RTMP_URL    "rtmp://127.0.0.1:1935/live/01"
#define PUSH_FILE   "./test.h264"
#define HTTP_URL    "http://127.0.0.1:8080/live/01.flv"





#define   GBUFF_MAX_SIZE 10*1024 * 1024
#define   ONE_READ_SIZE 1024*1024

char gbuf[GBUFF_MAX_SIZE] = { 0 };
int  gbufDatasize = 0;
int  gBytesRead = 0;

#define   BLOCKBUFF_MAX_SIZE 10*1024 * 1024


int TestRtmpPublisher(xop::EventLoop *event_loop);
char * sdPath = "data/data/com.example.testrtsp/1218.h264";

CGBlockRingBuffer*  pRingBuffer = NULL;

int rtmpMain()
{
    LOGE("rtmpMain---->");
    pRingBuffer = new CGBlockRingBuffer();
    pRingBuffer->Init(BLOCKBUFF_MAX_SIZE);
    //WriteGbufToFile();
    //return 0;

    int count = 1; 
#if TEST_MULTI_THREAD
    count = std::thread::hardware_concurrency();
#endif
    xop::EventLoop event_loop(count);

    /* rtmp server example */
    auto rtmp_server = xop::RtmpServer::Create(&event_loop);
    rtmp_server->SetChunkSize(60000);
    //rtmp_server->SetGopCache(); /* enable gop cache */
    rtmp_server->SetEventCallback([](std::string type, std::string stream_path) {
        LOGE("[Event] %s, stream path: %s\n\n", type.c_str(), stream_path.c_str());
    });
    if (!rtmp_server->Start("0.0.0.0", 1935)) {
        LOGE("RTMP Server listen on 1935 failed.\n");
    }

    /* http-flv server example */
    /*
    xop::HttpFlvServer http_flv_server; 
    http_flv_server.Attach(rtmp_server);
    if (!http_flv_server.Start("0.0.0.0", 8080)) {
        printf("HTTP FLV Server listen on 8080 failed.\n");
    }*/

#if TEST_RTMP_PUSHER
    /* rtmp pusher example */
    std::thread t([&event_loop] () {
        TestRtmpPublisher(&event_loop);
    });
    t.detach();
#endif 
/*
#if    TEST_RTMP_CLIENT
    auto rtmp_client = xop::RtmpClient::Create(&event_loop);
    rtmp_client->SetFrameCB([](uint8_t* payload, uint32_t length, uint8_t codecId, uint32_t timestamp) {
        printf("recv frame, type:%u, size:%u,\n", codecId, length);
    });

    std::string status;
    if (rtmp_client->OpenUrl(RTMP_URL, 3000, status) != 0) {
        printf("Open url %s failed, status: %s\n", RTMP_URL, status.c_str());
    }
#endif
*/

    while (1) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    
    rtmp_server->Stop();
    //http_flv_server.Stop();
    return 0;
}


class H264File
{
public:
    H264File(int bufSize = 5000000);
    ~H264File();

    bool open(const char *path);
    void Close();

    bool isOpened() const
    {
        return (m_file != NULL);
    }

    int readFrame(char *inBuf, int inBufSize, bool *bEndOfFrame);
    int readFrameFromGbuf(char* inBuf, int inBufSize);
    int readfileToGBbuf();
private:
    FILE *m_file = NULL;
    char *m_buf = NULL;
    int m_bufSize = 0;
    int m_bytesUsed = 0;
    int m_count = 0;
};

H264File::H264File(int bufSize)
    : m_bufSize(bufSize)
{
    m_buf = new char[m_bufSize];
}

H264File::~H264File()
{
    delete m_buf;
}

bool H264File::open(const char *path)
{
    m_file = fopen(path, "rb");
    if (m_file == NULL)
    {
        return false;
    }

    return true;
}

void H264File::Close()
{
    if (m_file)
    {
        fclose(m_file);
        m_file = NULL;
        m_count = 0;
        m_bytesUsed = 0;
    }
}

/*
int H264File::readfileToGBbuf() {
    fseek(m_file, 0, SEEK_SET);
    //fseek(m_file, 31, SEEK_SET);
    int bytesRead = (int)fread(gbuf, 1, GBUFF_MAX_SIZE, m_file);
    gbufDatasize = bytesRead;
    fclose(m_file);
    printf("readfileToGBbuf ----->\n");
    return 0;
}*/

 int   pushOneFrame(char* inBuf, int inBufSize){
     /*
     if(gbufDatasize+inBufSize > 5*1024*1024) {
         LOGE("pushOneFrame out---- %d--->\n",gbufDatasize);
         return -1;
     }
      memcpy(gbuf+ gbufDatasize, inBuf, inBufSize);
      gbufDatasize += inBufSize;
      */

      pRingBuffer->Write((unsigned char*)inBuf,inBufSize);
      return 1;
 }

 //unsigned char tmpBuffer[1024*1024] = {0};

int H264File::readFrameFromGbuf(char* inBuf, int inBufSize) {
    bool b;
    bool* bEndOfFrame = &b;

/*
    int tmpOneStepRead = ONE_READ_SIZE;
    if (gBytesRead >= gbufDatasize) {
        gBytesRead = 0;
        LOGE("read over----->");
        return -1;
    }
    LOGE("readFrameFromGbuf %d",gBytesRead ,gbufDatasize);
    LOGE("read %d  size %d",inBufSize);
    if (gbufDatasize - gBytesRead < ONE_READ_SIZE)
        tmpOneStepRead = gbufDatasize - gBytesRead;
*/
    int tmpOneStepRead = pRingBuffer->getNotRead((unsigned char*)m_buf,inBufSize);

   // memcpy(m_buf, gbuf+ gBytesRead, tmpOneStepRead);
    LOGE("----%d,%d",tmpOneStepRead,inBufSize);

    int bytesRead = tmpOneStepRead;
    bool bFindStart = false, bFindEnd = false;

    int i = 0, startCode = 3;
    *bEndOfFrame = false;
    for (i = 0; i < bytesRead - 5; i++)
    {
        if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 1)
        {
            startCode = 3;
        }
        else if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 0 && m_buf[i + 3] == 1)
        {
            startCode = 4;
        }
        else
        {
            continue;
        }

        if (((m_buf[i + startCode] & 0x1F) == 0x5 || (m_buf[i + startCode] & 0x1F) == 0x1) &&
            ((m_buf[i + startCode + 1] & 0x80) == 0x80))
        {
            bFindStart = true;
            i += 4;
            break;
        }
    }

    for (; i < bytesRead - 5; i++)
    {
        if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 1)
        {
            startCode = 3;
        }
        else if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 0 && m_buf[i + 3] == 1)
        {
            startCode = 4;
        }
        else
        {
            continue;
        }

        if (((m_buf[i + startCode] & 0x1F) == 0x7) || ((m_buf[i + startCode] & 0x1F) == 0x8)
            || ((m_buf[i + startCode] & 0x1F) == 0x6) || (((m_buf[i + startCode] & 0x1F) == 0x5
                || (m_buf[i + startCode] & 0x1F) == 0x1) && ((m_buf[i + startCode + 1] & 0x80) == 0x80)))
        {
            bFindEnd = true;
            break;
        }
    }

    bool flag = false;
    if (bFindStart && !bFindEnd && m_count > 0)
    {
        flag = bFindEnd = true;
        i = bytesRead;
        *bEndOfFrame = true;
    }

    if (!bFindStart || !bFindEnd)
    {
        LOGE("!not find");
        this->Close();
        return -1;
    }

    int size = (i <= inBufSize ? i : inBufSize);
   // memcpy(inBuf, m_buf, size);
    int ret = pRingBuffer->Read((unsigned char*)inBuf,size);
    if(ret != size){
        LOGE("ret %d != size %d\n",ret,size);
    }

    if (!flag)
    {
        m_count += 1;
        m_bytesUsed += i;
    }
    else
    {
        m_count = 0;
        m_bytesUsed = 0;
    }

    //fseek(m_file, m_bytesUsed, SEEK_SET);
   // gBytesRead = m_bytesUsed;

    return size;
}



int TestRtmpPublisher(xop::EventLoop *event_loop)
{
    H264File h264_file;
    /*
    if (!h264_file.open(PUSH_FILE)) {
        LOGE("Open %s failed.\n", PUSH_FILE);
        return -1;
    }
    */
    /* push stream to local rtmp server */
    xop::MediaInfo media_info;
    auto publisher = xop::RtmpPublisher::Create(event_loop);
    publisher->SetChunkSize(60000);

    std::string status;
    if (publisher->OpenUrl(RTMP_URL, 3000, status) < 0) {
        LOGE("Open url %s failed, status: %s\n", RTMP_URL, status.c_str());
        return -1;
    }

    int buf_size = 500000;
    bool end_of_frame = false;
    bool has_sps_pps = false;
    uint8_t *frame_buf = new uint8_t[buf_size];
    //h264_file.readfileToGBbuf();

    sleep(3);
    while (publisher->IsConnected()) 
    {
        //int frameSize = h264_file.readFrame((char*)frame_buf, buf_size, &end_of_frame);
        int frameSize = h264_file.readFrameFromGbuf((char*)frame_buf, buf_size);
        if (frameSize > 0) {
            if (!has_sps_pps) {
                if (frame_buf[3] == 0x67 || frame_buf[4] == 0x67) {
                    xop::Nal sps = xop::H264Parser::findNal(frame_buf, frameSize);
                    if (sps.first != nullptr && sps.second != nullptr && *sps.first == 0x67) {
                        media_info.sps_size = (uint32_t)(sps.second - sps.first + 1);
                        media_info.sps.reset(new uint8_t[media_info.sps_size], std::default_delete<uint8_t[]>());
                        memcpy(media_info.sps.get(), sps.first, media_info.sps_size);

                        xop::Nal pps = xop::H264Parser::findNal(sps.second, frameSize - (int)(sps.second - frame_buf));
                        if (pps.first != nullptr && pps.second != nullptr && *pps.first == 0x68) {
                            media_info.pps_size = (uint32_t)(pps.second - pps.first + 1);
                            media_info.pps.reset(new uint8_t[media_info.pps_size], std::default_delete<uint8_t[]>());
                            memcpy(media_info.pps.get(), pps.first, media_info.pps_size);

                            has_sps_pps = true;
                            publisher->SetMediaInfo(media_info); /* set sps pps */                            
                            LOGE("Start rtmp pusher, rtmp url: %s , http-flv url: %s \n\n", RTMP_URL, HTTP_URL);
                        }
                    }
                }
            }
            
            if (has_sps_pps) {
                publisher->PushVideoFrame(frame_buf, frameSize); /* send h.264 frame */
            }
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(40));
    }
    
    delete frame_buf;
    return 0;
}
/*
void WriteGbufToFile(){
    FILE *m_file = NULL;
    m_file = fopen(sdPath, "wb");
    if (m_file == NULL)
    {
        LOGE("Open %s error %d!",sdPath,errno);
        return ;
    }    
    int ret = fwrite(gbuf,gbufDatasize,1,m_file);
    LOGE("ret %d --->!",ret);
      fclose(m_file);
    return;
}
*/

/*

int H264File::readFrame(char *inBuf, int inBufSize, bool *bEndOfFrame)
{
    printf("readFrame-------------------------->\n");
    if (m_file == NULL)
    {
        return -1;
    }

    int bytesRead = (int)fread(m_buf, 1, m_bufSize, m_file);
    if (bytesRead == 0)
    {
        fseek(m_file, 0, SEEK_SET);
        m_count = 0;
        m_bytesUsed = 0;
        bytesRead = (int)fread(m_buf, 1, m_bufSize, m_file);
        if (bytesRead == 0)
        {
            this->Close();
            return -1;
        }
    }

    bool bFindStart = false, bFindEnd = false;

    int i = 0, startCode = 3;
    *bEndOfFrame = false;
    for (i = 0; i < bytesRead - 5; i++)
    {
        if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 1)
        {
            startCode = 3;
        }
        else if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 0 && m_buf[i + 3] == 1)
        {
            startCode = 4;
        }
        else
        {
            continue;
        }

        if (((m_buf[i + startCode] & 0x1F) == 0x5 || (m_buf[i + startCode] & 0x1F) == 0x1) &&
            ((m_buf[i + startCode + 1] & 0x80) == 0x80))
        {
            bFindStart = true;
            i += 4;
            break;
        }
    }

    for (; i < bytesRead - 5; i++)
    {
        if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 1)
        {
            startCode = 3;
        }
        else if (m_buf[i] == 0 && m_buf[i + 1] == 0 && m_buf[i + 2] == 0 && m_buf[i + 3] == 1)
        {
            startCode = 4;
        }
        else
        {
            continue;
        }

        if (((m_buf[i + startCode] & 0x1F) == 0x7) || ((m_buf[i + startCode] & 0x1F) == 0x8)
            || ((m_buf[i + startCode] & 0x1F) == 0x6) || (((m_buf[i + startCode] & 0x1F) == 0x5
                || (m_buf[i + startCode] & 0x1F) == 0x1) && ((m_buf[i + startCode + 1] & 0x80) == 0x80)))
        {
            bFindEnd = true;
            break;
        }
    }

    bool flag = false;
    if (bFindStart && !bFindEnd && m_count > 0)
    {
        flag = bFindEnd = true;
        i = bytesRead;
        *bEndOfFrame = true;
    }

    if (!bFindStart || !bFindEnd)
    {
        this->Close();
        return -1;
    }

    int size = (i <= inBufSize ? i : inBufSize);
    memcpy(inBuf, m_buf, size);

    if (!flag)
    {
        m_count += 1;
        m_bytesUsed += i;
    }
    else
    {
        m_count = 0;
        m_bytesUsed = 0;
    }

    fseek(m_file, m_bytesUsed, SEEK_SET);
    return size;
}


*/