update

CMakeLists.txt

@@ -5,7 +5,7 @@ include(ExternalProject)
 
 project(SharedMemoryProtocol VERSION 0.1 DESCRIPTION "...")
 
-set(CMAKE_CXX_FLAGS "${CMAKE_C_FLAGS} -g -I..")
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g -I..")
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 set(ROOT .)
 set(INCLUDE ${ROOT}/src ${ROOT}/lib/include)
@@ -58,7 +58,7 @@ add_custom_command(TARGET unittest PRE_BUILD
 
 add_test(
   NAME UNITTEST_DIRECTLY
-  COMMAND ./unittest)
+  COMMAND valgrind --leak-check=full --log-fil=valgrind-log.txt ./unittest)
 
 # Link Google Test
 target_link_libraries(unittest gtest_main)

src/ioctx.cc

 #include "ioctx.h"
+#include <iostream>
 
 namespace IOCtx {
 
@@ -46,7 +47,7 @@ void InCtx::readFrame(AVPacket *packet) {
 ///////////////////////////////////////////////////////////////////////////////
 
 void OutCtx::writeFrame(AVPacket *packet) {
-    if (av_write_frame(fmt.get(), packet) < 0) {
+    if (av_write_frame(static_cast<AVFormatContext*>(fmt.get()), packet) < 0) {
         throw IO_ERROR();
     }
 }
@@ -57,7 +58,12 @@ IOCTX_ERROR OutCtx::newStream(AVCodecParameters* par) {
     }
 
     AVStream *s = avformat_new_stream(fmt.get(), nullptr);
-    avcodec_parameters_copy(s->codecpar, par);
+    if (s == nullptr) {
+        return ERROR;
+    }
+    if (avcodec_parameters_copy(s->codecpar, par) < 0) {
+        return ERROR;
+    }
 
     return OK;
 }
@@ -66,8 +72,11 @@ void OutCtx::writeHeader() {
     if (!isCustomIO && avio_open(
             &fmt.get()->pb, path.c_str(), AVIO_FLAG_WRITE) < 0)
         throw FAILED_TO_WRITE_HEADER();
-    if (avformat_write_header(fmt.get(), nullptr) < 0)
+
+    AVDictionary *opts = nullptr;
+    if (avformat_write_header(fmt.get(), &opts) < 0) {
         throw FAILED_TO_WRITE_HEADER();
+    }
 }
 
 void OutCtx::writeTrailer() {

src/ioctx.h

@@ -21,17 +21,17 @@ class OutCtx;
 
 class END_OF_FILE: public std::runtime_error {
 public:
-    END_OF_FILE(): std::runtime_error("") {}
+    END_OF_FILE(): std::runtime_error("END_OF_FILE") {}
 };
 
 class IO_ERROR: public std::runtime_error {
 public:
-    IO_ERROR(): std::runtime_error("") {}
+    IO_ERROR(): std::runtime_error("IO_ERROR") {}
 };
 
 class FAILED_TO_WRITE_HEADER: public std::runtime_error {
 public:
-    FAILED_TO_WRITE_HEADER(): std::runtime_error("") {}
+    FAILED_TO_WRITE_HEADER(): std::runtime_error("FAILED_TO_WRITE_HEADER") {}
 };
 
 enum IOCTX_ERROR {
@@ -44,8 +44,9 @@ using StreamPrediction = std::function<bool(AVStream*)>;
 
 class InOutCtx {
 public:
+    InOutCtx() = default;
     InOutCtx(std::string path, bool isCustom):
-        fmt(nullptr), path(path), isCustomIO(false) {}
+        fmt(nullptr), path(path), isCustomIO(isCustom) {}
 
     AVStream* getStream(StreamPrediction);
     bool isReady() noexcept;
@@ -53,6 +54,7 @@ protected:
     Utils::AVFormatContextShared fmt;
     std::string path;
     bool isCustomIO;
+    std::shared_ptr<void> customProto;
 };
 
 
@@ -61,6 +63,7 @@ class InCtx: public InOutCtx {
     // to initialize.
     friend OutCtx;
 public:
+    InCtx() = default;
     InCtx(std::string path): InOutCtx(path, false) {
         fmt = Utils::makeInAVFormat(path, nullptr);
     }
@@ -69,8 +72,11 @@ public:
     InCtx(std::string path, IOProto::IOProtocol<T> *proto):
         InOutCtx(path, true) {
 
+        customProto = std::shared_ptr<IOProto::IOProtocol<T>>(proto);
+
         // FIXME: AVIOContext should be generated from proto
-        AVIOContext *ioctx = proto.to_avioctx();
+        AVIOContext *ioctx = proto->to_avioctx();
+
         fmt = Utils::makeInAVFormat(path, ioctx);
     }
 
@@ -80,18 +86,20 @@ public:
 
 class OutCtx: public InOutCtx {
 public:
+    OutCtx() = default;
     OutCtx(std::string path): InOutCtx(path, false) {
-        fmt = Utils::makeOutAVFormat(path, nullptr);
+        fmt = Utils::makeOutAVFormat(path, nullptr, nullptr);
     }
 
     template<typename T>
-    OutCtx(std::string path, IOProto::IOProtocol<T> *proto):
+    OutCtx(std::string path, IOProto::IOProtocol<T> *proto, AVOutputFormat *ofmt):
         InOutCtx(path, true) {
 
-        // FIXME: AVIOContext should be generated from proto
-        AVIOContext *ioctx = proto.to_avioctx();
+        customProto = std::shared_ptr<IOProto::IOProtocol<T>>(proto);
 
-        fmt = Utils::makeOutAVFormat(path, ioctx);
+        // FIXME: AVIOContext should be generated from proto
+        AVIOContext *ioctx = proto->to_avioctx();
+        fmt = Utils::makeOutAVFormat(path, ioctx, ofmt);
     }
 
     void writeFrame(AVPacket*);

src/proto/proto.h

@@ -13,7 +13,7 @@ extern "C" {
 namespace IOProto {
 
 
-constexpr size_t DEFAULT_BUFFER_SIZE = 1 << 15;
+constexpr size_t DEFAULT_BUFFER_SIZE = 32768;
 
 enum RW_FLAG {
     read,
@@ -43,11 +43,22 @@ public:
     AVIOContext* to_avioctx() noexcept {
 
         if (io == nullptr) {
-            buffer = std::make_unique<uint8_t>(DEFAULT_BUFFER_SIZE);
-            io = avio_alloc_context(buffer.get(), DEFAULT_BUFFER_SIZE, flag, priv,
-                                    this->read_packet,
-                                    this->write_packet,
-                                    this->seek_packet);
+
+            // This buffer is managed by libav
+            uint8_t *buffer = new uint8_t[DEFAULT_BUFFER_SIZE];
+
+            io = avio_alloc_context(
+                buffer, DEFAULT_BUFFER_SIZE, flag == write, static_cast<void*>(this),
+                [](void *priv, uint8_t *buf, int bufSize) -> int {
+                    return static_cast<T*>(priv)->read_packet(priv, buf, bufSize);
+                },
+                [](void *priv, uint8_t *buf, int bufSize) -> int {
+                    return static_cast<T*>(priv)->write_packet(priv, buf, bufSize);
+                },
+                [](void *priv, int64_t offset, int whence) -> int64_t {
+                    return static_cast<T*>(priv)->seek_packet(priv, offset, whence);
+                });
+            io->direct = true;
         }
 
         return io;
@@ -58,7 +69,6 @@ protected:
     void *priv;
     std::string name;
     AVIOContext *io;
-    std::unique_ptr<uint8_t> buffer;
 };
 
 }

src/proto/transientMemProto.cc

 #include "transientMemProto.h"
 #include <memory>
 #include <algorithm>
+#include <queue>
+
+extern "C" {
+    #include <malloc.h>
+}
+
+using std::queue;
 
 namespace IOProto {
 namespace TransientMemProto {
 
 /* External buffer must not be realeased
-    * by TransientMemProto */
-static uint8_t *externalBuffer;
-static size_t   externalBufferSize;
-
-static uint8_t *bufPtr;
-static uint8_t *bufEnd;
-static size_t   dataSize;
+ * by TransientMemProto */
+struct BufferInfo {
+    uint8_t *buffer;
+    uint8_t *pos;
+    size_t   remain;
+    bool     eof;
+};
+
+BufferInfo        current;
+queue<BufferInfo> externalBuffers;
+
+// Configurations
 static bool     isRleaseExtBuf = false;
+static bool     debugging      = true;
+
 
-bool isEmpty() {
-    return dataSize == 0;
-}
 
 void config_releaseExtBuf(bool onoff) {
     isRleaseExtBuf = onoff;
@@ -25,14 +36,7 @@ void config_releaseExtBuf(bool onoff) {
 
 /* Note: The passed buf should be full filled */
 void attachBuffer(uint8_t *buf, size_t bufSize) {
-    if (externalBuffer && isRleaseExtBuf)
-        free(externalBuffer);
-
-    externalBuffer = buf;
-    externalBufferSize = bufSize;
-
-    bufPtr = buf;
-    bufEnd = buf + bufSize;
+    externalBuffers.push({ buf, buf, bufSize, false });
 }
 
 int TransientMemProto::read_packet_internal(void *priv, uint8_t *buf, int bufSize) {
@@ -40,25 +44,42 @@ int TransientMemProto::read_packet_internal(void *priv, uint8_t *buf, int bufSiz
         AVERROR(EINVAL);
     }
 
-    size_t readSize = std::min((size_t)bufSize, dataSize);
+    // No Datas
+    if (current.remain == 0 && externalBuffers.empty()) {
+        return AVERROR(EAGAIN);
+    }
+
+    if (current.remain == 0) {
+        current = externalBuffers.front();
+        externalBuffers.pop();
+
+        if (current.eof) return AVERROR_EOF;
+    }
+
+    size_t readSize = std::min((size_t)bufSize, current.remain);
 
     // Read data
-    memcpy(buf, bufPtr, readSize);
+    memcpy(buf, current.pos, readSize);
 
     // Update Buffer status
-    bufPtr   += readSize;
-    dataSize -= readSize;
+    current.pos += readSize;
+    current.remain -= readSize;
 
-    return readSize;
+    if (current.remain == 0 && isRleaseExtBuf) {
+        delete current.buffer;
+    }
+
+    return readSize == 0 ? AVERROR(EAGAIN) : readSize;
 }
 
 int TransientMemProto::write_packet_internal(void *priv, uint8_t *buf, int bufSize) {
-    uint8_t *newBuf = new uint8_t(bufSize);
+
+    uint8_t *newBuf = new uint8_t[bufSize];
 
     memcpy(newBuf, buf, bufSize);
     attachBuffer(newBuf, bufSize);
 
-    return 0;
+    return bufSize;
 }
 
 int64_t TransientMemProto::seek_packet_internal(void *opaque, int64_t offset, int whence) {

src/utils.cc

 #include "utils.h"
+#include <iostream>
 
 namespace Utils {
 
@@ -37,19 +38,21 @@ AVFormatContextShared makeInAVFormat(std::string path, AVIOContext *customIO) {
 }
 
 static AVFormatContext* AVFormatOutputContextConstructor(
-    std::string path, AVIOContext *customIO) {
+    std::string path, AVIOContext *customIO, AVOutputFormat *customFormat) {
 
     AVFormatContext *ctx = nullptr;
 
+    if (customFormat != nullptr)
+        path = "";
+
+    if (avformat_alloc_output_context2(&ctx, customFormat, nullptr, path.c_str()) < 0) {
+        return nullptr;
+    }
+
     if (customIO != nullptr) {
-        ctx = avformat_alloc_context();
         ctx->pb = customIO;
-        path = path == "" ? path : "";
     }
 
-    if (avformat_alloc_output_context2(&ctx, nullptr, nullptr, path.c_str()) < 0) {
-        return nullptr;
-    }
 
     return ctx;
 }
@@ -58,9 +61,11 @@ static void AVFormatOutputContextDestructor(AVFormatContext *ctx) {
     avformat_close_input(&ctx);
 }
 
-AVFormatContextShared makeOutAVFormat(std::string path, AVIOContext *customIO) {
+AVFormatContextShared makeOutAVFormat(
+    std::string path, AVIOContext *customIO, AVOutputFormat *customFormat) {
+
     AVFormatContextShared ioCtx {
-        AVFormatOutputContextConstructor(path, customIO),
+        AVFormatOutputContextConstructor(path, customIO, customFormat),
         AVFormatOutputContextDestructor
     };
 

src/utils.h

@@ -12,7 +12,7 @@ namespace Utils {
 
 using AVFormatContextShared = std::shared_ptr<AVFormatContext>;
 AVFormatContextShared makeInAVFormat(std::string, AVIOContext*);
-AVFormatContextShared makeOutAVFormat(std::string, AVIOContext*);
+AVFormatContextShared makeOutAVFormat(std::string, AVIOContext*, AVOutputFormat*);
 
 /* To check that is video specified by path valid
  *

tests/ioctxTestCases.cc

@@ -2,12 +2,15 @@
 #include <string>
 #include "ioctx.h"
 #include "utils.h"
+#include "proto/transientMemProto.h"
 
 extern "C" {
 #include <libavformat/avformat.h>
 #include <libavutil/avutil.h>
+#include <malloc.h>
 }
 
+namespace TrMemProto = IOProto::TransientMemProto;
 
 class IOCTX_With_Default_Proto_Fixture: public ::testing::Test {
 protected:
@@ -38,12 +41,30 @@ protected:
 class IOCTX_With_TransientMem_Proto_Fixture: public ::testing::Test {
 protected:
     void SetUp() override {
+        TrMemProto::TransientMemProto *outProto =
+            new TrMemProto::TransientMemProto(nullptr, IOProto::write);
+        AVOutputFormat *h264Output = av_guess_format("h264", nullptr, nullptr);
+        outCtxCustom = new IOCtx::OutCtx("", outProto, h264Output);
 
+        AVStream *is = iCtx.getStream([](AVStream *s) {
+            return s->codecpar->codec_type == AVMEDIA_TYPE_VIDEO;
+        });
+        if (outCtxCustom->newStream(is->codecpar) == IOCtx::ERROR) {
+            throw std::runtime_error("SetUp Failed");
+        }
+        outCtxCustom->writeHeader();
+        if (oCtx.newStream(is->codecpar) == IOCtx::ERROR)
+            throw std::runtime_error("Failed to init fixture");
+        oCtx.writeHeader();
     }
 
     std::string inFilePath = "./resources/small_bunny_1080p_60fps.mp4";
     std::string outFilePath = "./resources/small_bunny_1080p_60fps_out.mp4";
 
+    IOCtx::InCtx iCtx { inFilePath };
+    IOCtx::OutCtx *outCtxCustom;
+    IOCtx::InCtx *inCtxCustom;
+    IOCtx::OutCtx oCtx { outFilePath };
 };
 
 TEST_F(IOCTX_With_Default_Proto_Fixture, Description) {
@@ -73,4 +94,32 @@ TEST_F(IOCTX_With_Default_Proto_Fixture, Description) {
 
 TEST_F(IOCTX_With_TransientMem_Proto_Fixture, Description) {
 
+    AVPacket *packet = av_packet_alloc();
+    for (int i = 0; i < 200; ++i) {
+        iCtx.readFrame(packet);
+        outCtxCustom->writeFrame(packet);
+        av_packet_unref(packet);
+    }
+
+    TrMemProto::TransientMemProto *inProto =
+        new TrMemProto::TransientMemProto(nullptr, IOProto::read);
+    inCtxCustom = new IOCtx::InCtx("", inProto);
+
+
+    try{
+    while (true) {
+        iCtx.readFrame(packet);
+        outCtxCustom->writeFrame(packet);
+
+        // Make sure all buffered datas is used
+        inCtxCustom->readFrame(packet);
+
+        oCtx.writeFrame(packet);
+
+        av_packet_unref(packet);
+    }
+    } catch (IOCtx::END_OF_FILE e) {}
+
+    oCtx.writeTrailer();
+
 }