#29564, #29818: new video node based system.

Fix bug #29818.
Note: non-conference mode has been tested and works.
Conference mode may works, but some issues have been detected.

daemon/src/client/dbus/video_controls.cpp

@@ -182,7 +182,7 @@ VideoControls::stopPreview()
     }
 }
 
-sfl_video::VideoSource* VideoControls::getVideoPreview()
+sfl_video::VideoFrameActiveWriter* VideoControls::getVideoPreview()
 {
     return videoPreview_.get();
 }

daemon/src/client/video_controls.h

@@ -57,16 +57,13 @@
 
 #include <memory> // for shared_ptr
 #include "video/video_preferences.h"
-
-namespace sfl_video {
-    class VideoSource;
-}
+#include "video/video_base.h"
 
 class VideoControls : public org::sflphone::SFLphone::VideoControls_adaptor,
     public DBus::IntrospectableAdaptor,
     public DBus::ObjectAdaptor {
     private:
-        std::shared_ptr<sfl_video::VideoSource> videoPreview_;
+        std::unique_ptr<sfl_video::VideoFrameActiveWriter> videoPreview_;
         VideoPreference videoPreference_;
 
     public:
@@ -126,7 +123,7 @@ class VideoControls : public org::sflphone::SFLphone::VideoControls_adaptor,
         void startPreview();
         void stopPreview();
         bool hasPreviewStarted();
-		sfl_video::VideoSource* getVideoPreview();
+        sfl_video::VideoFrameActiveWriter* getVideoPreview();
 };
 
 #endif // VIDEO_CONTROLS_H_

daemon/src/conference.cpp

@@ -41,6 +41,8 @@
 #include "video/video_camera.h"
 #endif
 
+#include "logger.h"
+
 
 Conference::Conference()
     : id_(Manager::instance().getNewCallID())
@@ -51,15 +53,6 @@ Conference::Conference()
 #endif
 {
     Recordable::initRecFilename(id_);
-
-#ifdef SFL_VIDEO
-    sfl_video::VideoCamera *camera = static_cast<sfl_video::VideoCamera*>(Manager::instance().getVideoControls()->getVideoPreview());
-    if (camera) {
-        videoMixer_.addSource(camera);
-        camera->setMixer(&videoMixer_);
-    }
-#endif
-
 }
 
 Conference::ConferenceState Conference::getState() const
@@ -84,11 +77,16 @@ void Conference::remove(const std::string &participant_id)
 
 void Conference::bindParticipant(const std::string &participant_id)
 {
-    for (const auto &item : participants_)
+    auto mainBuffer = Manager::instance().getMainBuffer();
+
+    for (const auto &item : participants_) {
         if (participant_id != item)
-            Manager::instance().getMainBuffer().bindCallID(participant_id, item);
+            mainBuffer.bindCallID(participant_id, item);
+        mainBuffer.flush(item);
+    }
 
-    Manager::instance().getMainBuffer().bindCallID(participant_id, MainBuffer::DEFAULT_ID);
+    mainBuffer.bindCallID(participant_id, MainBuffer::DEFAULT_ID);
+    mainBuffer.flush(MainBuffer::DEFAULT_ID);
 }
 
 std::string Conference::getStateStr() const

daemon/src/managerimpl.cpp

@@ -930,12 +930,15 @@ ManagerImpl::addParticipant(const std::string& callId, const std::string& confer
     std::string callState(callDetails.find("CALL_STATE")->second);
 
     if (callState == "HOLD") {
+        ERROR("foo1: %s", callId.c_str());
         conf->bindParticipant(callId);
         offHoldCall(callId);
     } else if (callState == "INCOMING") {
+        ERROR("foo2: %s", callId.c_str());
         conf->bindParticipant(callId);
         answerCall(callId);
     } else if (callState == "CURRENT")
+        ERROR("foo3: %s", callId.c_str());
         conf->bindParticipant(callId);
 
     ParticipantSet participants(conf->getParticipantList());
@@ -943,13 +946,6 @@ ManagerImpl::addParticipant(const std::string& callId, const std::string& confer
     if (participants.empty())
         ERROR("Participant list is empty for this conference");
 
-    // reset ring buffer for all conference participant
-    // flush conference participants only
-    for (const auto &p : participants)
-        getMainBuffer().flush(p);
-
-    getMainBuffer().flush(MainBuffer::DEFAULT_ID);
-
     // Connect stream
     addStream(callId);
     return true;
@@ -1349,15 +1345,8 @@ void ManagerImpl::addStream(const std::string& call_id)
         if (iter != conferenceMap_.end() and iter->second) {
             Conference* conf = iter->second;
 
+            ERROR("bar: %s", call_id.c_str());
             conf->bindParticipant(call_id);
-
-            ParticipantSet participants(conf->getParticipantList());
-
-            // reset ring buffer for all conference participant
-            for (const auto &participant : participants)
-                getMainBuffer().flush(participant);
-
-            getMainBuffer().flush(MainBuffer::DEFAULT_ID);
         }
 
     } else {

daemon/src/video/shm_sink.cpp

@@ -37,10 +37,12 @@
 #include "config.h"
 #endif
 
+#include "video_scaler.h"
+
 #include "shm_sink.h"
 #include "shm_header.h"
 #include "logger.h"
-#include "video_provider.h"
+
 #include <sys/mman.h>
 #include <fcntl.h>
 #include <cstdio>
@@ -49,13 +51,15 @@
 #include <cerrno>
 #include <cstring>
 
+namespace sfl_video {
+
 SHMSink::SHMSink(const std::string &shm_name) :
-    shm_name_(shm_name),
-    fd_(-1),
-    shm_area_(static_cast<SHMHeader*>(MAP_FAILED)),
-    shm_area_len_(0),
-    opened_name_()
-    {}
+    shm_name_(shm_name)
+    , fd_(-1)
+    , shm_area_(static_cast<SHMHeader*>(MAP_FAILED))
+    , shm_area_len_(0)
+    , opened_name_()
+{}
 
 SHMSink::~SHMSink()
 {
@@ -187,8 +191,14 @@ void SHMSink::render(const std::vector<unsigned char> &data)
     shm_unlock();
 }
 
-void SHMSink::render_callback(sfl_video::VideoProvider &provider, size_t bytes)
+void SHMSink::render_frame(VideoFrame& src)
 {
+    VideoFrame dst;
+    VideoScaler scaler;
+
+    dst.setGeometry(src.getWidth(), src.getHeight(), VIDEO_PIXFMT_BGRA);
+    size_t bytes = dst.getSize();
+
     shm_lock();
 
     if (!resize_area(sizeof(SHMHeader) + bytes)) {
@@ -196,19 +206,38 @@ void SHMSink::render_callback(sfl_video::VideoProvider &provider, size_t bytes)
         return;
     }
 
-    provider.fillBuffer(static_cast<void*>(shm_area_->data));
+    dst.setDestination(shm_area_->data);
+    scaler.scale(src, dst);
+
     shm_area_->buffer_size = bytes;
     shm_area_->buffer_gen++;
     sem_post(&shm_area_->notification);
     shm_unlock();
 }
 
-void SHMSink::shm_lock()
+void SHMSink::render_callback(VideoProvider &provider, size_t bytes)
 {
-    sem_wait(&shm_area_->mutex);
+    shm_lock();
+
+    if (!resize_area(sizeof(SHMHeader) + bytes)) {
+        ERROR("Could not resize area");
+        return;
+    }
+
+    provider.fillBuffer(static_cast<void*>(shm_area_->data));
+    shm_area_->buffer_size = bytes;
+    shm_area_->buffer_gen++;
+    sem_post(&shm_area_->notification);
+    shm_unlock();
 }
 
+void SHMSink::shm_lock()
+{ sem_wait(&shm_area_->mutex); }
+
 void SHMSink::shm_unlock()
-{
-    sem_post(&shm_area_->mutex);
+{ sem_post(&shm_area_->mutex); }
+
+void SHMSink::update(Observable<VideoFrameSP>* obs, VideoFrameSP& frame_p)
+{ render_frame(*frame_p); }
+
 }

daemon/src/video/shm_sink.h

@@ -36,17 +36,20 @@
 #ifndef SHM_SINK_H_
 #define SHM_SINK_H_
 
+#include "noncopyable.h"
+#include "video_provider.h"
+#include "video_base.h"
+
 #include <string>
 #include <vector>
-#include "noncopyable.h"
 
 class SHMHeader;
+
 namespace sfl_video {
-    class VideoProvider;
-}
 
-class SHMSink {
-    public:
+class SHMSink : public VideoFramePassiveReader
+{
+public:
     SHMSink(const std::string &shm_name = "");
     std::string openedName() const { return opened_name_; }
     ~SHMSink();
@@ -57,9 +60,13 @@ class SHMSink {
     bool resize_area(size_t desired_length);
 
     void render(const std::vector<unsigned char> &data);
-        void render_callback(sfl_video::VideoProvider &provider, size_t bytes);
+    void render_frame(VideoFrame& src);
+    void render_callback(VideoProvider &provider, size_t bytes);
+
+    // as VideoFramePassiveReader
+    void update(Observable<VideoFrameSP>*, VideoFrameSP&);
 
-    private:
+private:
     NON_COPYABLE(SHMSink);
 
     void shm_lock();
@@ -71,4 +78,6 @@ class SHMSink {
     std::string opened_name_;
 };
 
+}
+
 #endif // SHM_SINK_H_

daemon/src/video/socket_pair.cpp

@@ -222,8 +222,10 @@ int SocketPair::readCallback(void *opaque, uint8_t *buf, int buf_size)
 						   {context->rtcpHandle_, POLLIN, 0}};
 
     for(;;) {
-        if (context->interrupted_)
+        if (context->interrupted_) {
+            ERROR("interrupted");
             return -EINTR;
+        }
 
         /* build fdset to listen to RTP and RTCP packets */
         n = poll(p, 2, 100);

daemon/src/video/video_base.cpp

@@ -82,12 +82,14 @@ VideoFrame::~VideoFrame()
     avcodec_free_frame(&frame_);
 }
 
-int VideoFrame::getFormat() const
-{
-    return libav_utils::sfl_pixel_format(frame_->format);
-}
-int VideoFrame::getWidth() const { return frame_->width; }
-int VideoFrame::getHeight() const { return frame_->height; }
+int VideoFrame::getPixelFormat() const
+{ return libav_utils::sfl_pixel_format(frame_->format); }
+
+int VideoFrame::getWidth() const
+{ return frame_->width; }
+
+int VideoFrame::getHeight() const
+{ return frame_->height; }
 
 bool VideoFrame::allocBuffer(int width, int height, int pix_fmt)
 {
@@ -207,70 +209,24 @@ void VideoFrame::test()
 
 /*=== VideoGenerator =========================================================*/
 
-VideoGenerator::VideoGenerator() :
-    VideoSource::VideoSource()
-    , mutex_()
-    , condition_()
-    , writableFrame_()
-    , lastFrame_()
-{
-    pthread_mutex_init(&mutex_, NULL);
-    pthread_cond_init(&condition_, NULL);
-}
-
-VideoGenerator::~VideoGenerator()
+VideoFrame& VideoGenerator::getNewFrame()
 {
-    pthread_cond_destroy(&condition_);
-    pthread_mutex_destroy(&mutex_);
+    if (writableFrame_)
+        writableFrame_->setdefaults();
+    else
+        writableFrame_.reset(new VideoFrame());
+    return *writableFrame_.get();
 }
 
 void VideoGenerator::publishFrame()
 {
-    pthread_mutex_lock(&mutex_);
-    {
-        lastFrame_ = std::move(writableFrame_); // we owns it now
-        pthread_cond_signal(&condition_);
-    }
-    pthread_mutex_unlock(&mutex_);
-}
-
-std::shared_ptr<VideoFrame> VideoGenerator::waitNewFrame()
-{
-    pthread_mutex_lock(&mutex_);
-    pthread_cond_wait(&condition_, &mutex_);
-    pthread_mutex_unlock(&mutex_);
-
-    return obtainLastFrame();
+    lastFrame_ = std::move(writableFrame_);
+    notify(std::ref(lastFrame_));
 }
 
-std::shared_ptr<VideoFrame> VideoGenerator::obtainLastFrame()
+VideoFrameSP VideoGenerator::obtainLastFrame()
 {
-    std::shared_ptr<VideoFrame> frame;
-
-    pthread_mutex_lock(&mutex_);
-    frame = lastFrame_;
-    pthread_mutex_unlock(&mutex_);
-
-    return frame;
-}
-
-VideoFrame& VideoGenerator::getNewFrame()
-{
-    VideoFrame* frame;
-
-    pthread_mutex_lock(&mutex_);
-    {
-        if (writableFrame_) {
-            frame = writableFrame_.get();
-            frame->setdefaults();
-        } else {
-            frame = new VideoFrame();
-            writableFrame_.reset(frame);
-        }
-    }
-    pthread_mutex_unlock(&mutex_);
-
-    return *frame;
+    return lastFrame_;
 }
 
 }

daemon/src/video/video_base.h

@@ -37,7 +37,9 @@
 #include <cstdlib>
 #include <cstdint>
 #include <memory>
-#include <forward_list>
+#include <set>
+#include <mutex>
+#include <condition_variable>
 
 
 class AVFrame;
@@ -55,10 +57,59 @@ enum VideoPixelFormat {
 
 namespace sfl_video {
 
+template <typename T> class Observator;
+template <typename T> class Observable;
+class VideoFrame;
+
 typedef int(*io_readcallback)(void *opaque, uint8_t *buf, int buf_size);
 typedef int(*io_writecallback)(void *opaque, uint8_t *buf, int buf_size);
 typedef int64_t(*io_seekcallback)(void *opaque, int64_t offset, int whence);
 
+typedef std::shared_ptr<VideoFrame> VideoFrameUP;
+typedef std::shared_ptr<VideoFrame> VideoFrameSP;
+
+/*=== Observable =============================================================*/
+
+template <typename T>
+class Observable
+{
+public:
+    Observable() : observators_(), mutex_() {}
+    virtual ~Observable() {};
+
+    void attach(Observator<T>* o) {
+        std::unique_lock<std::mutex> lk(mutex_);
+        observators_.insert(o);
+    }
+
+    void detach(Observator<T>* o) {
+        std::unique_lock<std::mutex> lk(mutex_);
+        observators_.erase(o);
+    }
+
+    void notify(T& data) {
+        std::unique_lock<std::mutex> lk(mutex_);
+        for (auto observator : observators_)
+            observator->update(this, data);
+    }
+
+private:
+    NON_COPYABLE(Observable<T>);
+
+	std::set<Observator<T>*> observators_;
+    std::mutex mutex_; // lock observators_
+};
+
+/*=== Observator =============================================================*/
+
+template <typename T>
+class Observator
+{
+public:
+    virtual ~Observator() {};
+	virtual void update(Observable<T>*, T&) = 0;
+};
+
 /*=== VideoPacket  ===========================================================*/
 
 class VideoPacket {
@@ -115,7 +166,7 @@ public:
     ~VideoFrame();
 
     AVFrame* get() { return frame_; };
-    int getFormat() const;
+    int getPixelFormat() const;
     int getWidth() const;
     int getHeight() const;
     void setGeometry(int width, int height, int pix_fmt);
@@ -134,36 +185,42 @@ private:
     bool allocated_;
 };
 
-/*=== VideoSource ============================================================*/
+/*=== VideoNode ============================================================*/
 
-class VideoSource {
-public:
-    virtual ~VideoSource() {}
-    virtual std::shared_ptr<VideoFrame> waitNewFrame() = 0;
-    virtual std::shared_ptr<VideoFrame> obtainLastFrame() = 0;
-    virtual int getWidth() const = 0;
-    virtual int getHeight() const = 0;
-};
+class VideoNode { public: virtual ~VideoNode() {}; };
+typedef VideoNode VideoSource;
+
+class VideoFrameActiveWriter :
+        public Observable<VideoFrameSP>,
+        public VideoNode
+{};
+
+class VideoFramePassiveReader :
+        public Observator<VideoFrameSP>,
+        public VideoNode
+{};
 
 /*=== VideoGenerator =========================================================*/
 
-class VideoGenerator : public VideoSource {
+class VideoGenerator : public VideoFrameActiveWriter
+{
 public:
-    VideoGenerator();
-    virtual ~VideoGenerator();
+    VideoGenerator() : writableFrame_(), lastFrame_() {}
+
+    virtual int getWidth() const = 0;
+    virtual int getHeight() const = 0;
+    virtual int getPixelFormat() const = 0;
 
-    std::shared_ptr<VideoFrame> waitNewFrame();
-    std::shared_ptr<VideoFrame> obtainLastFrame();
+    VideoFrameSP obtainLastFrame();
 
 protected:
-    void publishFrame();
+    // getNewFrame and publishFrame must be called by the same thread only
     VideoFrame& getNewFrame();
+    void publishFrame();
 
 private:
-    pthread_mutex_t mutex_;
-    pthread_cond_t condition_;
-    std::unique_ptr<VideoFrame> writableFrame_;
-    std::shared_ptr<VideoFrame> lastFrame_;
+    VideoFrameUP writableFrame_;
+    VideoFrameSP lastFrame_;
 };
 
 }

daemon/src/video/video_camera.cpp

@@ -43,20 +43,14 @@ namespace sfl_video {
 using std::string;
 
 VideoCamera::VideoCamera(const std::map<std::string, std::string> &args) :
-    VideoSource::VideoSource()
+    VideoGenerator::VideoGenerator()
     , id_("local")
     , args_(args)
     , decoder_(0)
     , sink_()
-    , bufferSize_(0)
-    , previewWidth_(0)
-    , previewHeight_(0)
-    , scaler_()
-    , frame_()
-    , mixer_()
-{
-    start();
-}
+    , sinkWidth_(0)
+    , sinkHeight_(0)
+{ start(); }
 
 VideoCamera::~VideoCamera()
 {
@@ -96,41 +90,37 @@ bool VideoCamera::setup()
 
     /* Preview frame size? (defaults from decoder) */
     if (!args_["width"].empty())
-        previewWidth_ = atoi(args_["width"].c_str());
+        sinkWidth_ = atoi(args_["width"].c_str());
     else
-        previewWidth_ = decoder_->getWidth();
+        sinkWidth_ = decoder_->getWidth();
     if (!args_["height"].empty())
-        previewHeight_ = atoi(args_["height"].c_str());
+        sinkHeight_ = atoi(args_["height"].c_str());
     else
-        previewHeight_ = decoder_->getHeight();
+        sinkHeight_ = decoder_->getHeight();
 
-    /* Previewing setup */
+    /* Sink setup */
     EXIT_IF_FAIL(sink_.start(), "Cannot start shared memory sink");
+    Manager::instance().getVideoControls()->startedDecoding(id_,
+                                                            sink_.openedName(),
+                                                            sinkWidth_,
+                                                            sinkHeight_);
+    DEBUG("TX: shm sink <%s> started: size = %dx%d",
+          sink_.openedName().c_str(), sinkWidth_, sinkHeight_);
 
-    frame_.setGeometry(previewWidth_, previewHeight_, VIDEO_PIXFMT_BGRA);
-    bufferSize_ = frame_.getSize();
-    EXIT_IF_FAIL(bufferSize_ > 0, "Incorrect buffer size for decoding");
+    attach(&sink_);
 
-    string name = sink_.openedName();
-    Manager::instance().getVideoControls()->startedDecoding(id_, name,
-                                                            previewWidth_,
-                                                            previewHeight_);
-    DEBUG("TX: shm sink started with size %d, width %d and height %d",
-          bufferSize_, previewWidth_, previewHeight_);
     return true;
 }
 
 void VideoCamera::process()
-{
-    if (captureFrame() and isRunning())
-        renderFrame();
-}
+{ captureFrame(); }
 
 void VideoCamera::cleanup()
 {
-    delete decoder_;
     Manager::instance().getVideoControls()->stoppedDecoding(id_,
                                                             sink_.openedName());
+    detach(&sink_);
+    delete decoder_;
 }
 
 int VideoCamera::interruptCb(void *data)
@@ -141,7 +131,7 @@ int VideoCamera::interruptCb(void *data)
 
 bool VideoCamera::captureFrame()
 {
-    int ret = decoder_->decode();
+    int ret = decoder_->decode(getNewFrame());