plugins: add call handler

Change-Id: If5296e71d4979962f71443fb298891202c8d2afe

bin/jni/plugin_manager_interface.i

@@ -17,4 +17,7 @@ std::vector<std::string> listAvailablePlugins();
 std::vector<std::string> listLoadedPlugins();
 int installPlugin(const std::string& jplPath, bool force);
 int uninstallPlugin(const std::string& pluginRootPath);
+std::vector<std::string> listCallMediaHandlers();
+void toggleCallMediaHandler(const std::string& id, const bool toggle);
+std::map<std::string,std::string> getCallMediaHandlerDetails(const std::string& id);
 }

src/client/plugin_manager_interface.cpp

@@ -80,4 +80,16 @@ int installPlugin(const std::string& jplPath, bool force) {
 int uninstallPlugin(const std::string& pluginRootPath) {
     return jami::Manager::instance().getJamiPluginManager().uninstallPlugin(pluginRootPath);
 }
+
+std::vector<std::string> listCallMediaHandlers() {
+    return jami::Manager::instance().getJamiPluginManager().getCallServicesManager().listCallMediaHandlers();
+}
+
+void toggleCallMediaHandler(const std::string& id, const bool toggle) {
+    return jami::Manager::instance().getJamiPluginManager().getCallServicesManager().toggleCallMediaHandler(id, toggle);
+}
+
+std::map<std::string,std::string> getCallMediaHandlerDetails(const std::string& id) {
+    return jami::Manager::instance().getJamiPluginManager().getCallServicesManager().getCallMediaHandlerDetails(id);
+}
 }

src/dring/plugin_manager_interface.h

@@ -37,5 +37,8 @@ DRING_PUBLIC std::vector<std::string> listAvailablePlugins();
 DRING_PUBLIC std::vector<std::string> listLoadedPlugins();
 DRING_PUBLIC int installPlugin(const std::string& jplPath, bool force);
 DRING_PUBLIC int uninstallPlugin(const std::string& pluginRootPath);
+DRING_PUBLIC std::vector<std::string> listCallMediaHandlers();
+DRING_PUBLIC void toggleCallMediaHandler(const std::string& id, const bool toggle);
+DRING_PUBLIC std::map<std::string,std::string> getCallMediaHandlerDetails(const std::string& id);
 }
 

src/media/video/video_rtp_session.h

@@ -102,6 +102,14 @@ public:
     void initRecorder(std::shared_ptr<MediaRecorder>& rec) override;
     void deinitRecorder(std::shared_ptr<MediaRecorder>& rec) override;
 
+    std::shared_ptr<VideoFrameActiveWriter>& getVideoLocal() {
+        return videoLocal_;
+    }
+
+    std::unique_ptr<VideoReceiveThread>& getVideoReceive() {
+        return receiveThread_;
+    }
+
 private:
     void setupConferenceVideoPipeline(Conference& conference);
     void setupVideoPipeline();

src/observer.h

@@ -27,9 +27,11 @@
 #include <cstdint>
 #include <memory>
 #include <set>
+#include <list>
 #include <mutex>
 #include <functional>
 #include <ciso646> // fix windows compiler bug
+#include "logger.h"
 
 namespace jami {
 
@@ -44,11 +46,23 @@ class Observable
 public:
     Observable() : mutex_(), observers_() {}
 
+    /**
+     * @brief ~Observable
+     * Detach all observers to avoid making them call this observable when
+     * destroyed
+     */
     virtual ~Observable() {
         std::lock_guard<std::mutex> lk(mutex_);
+
+        for(auto& pobs: priority_observers_) {
+            if(auto so = pobs.lock()) {
+                so->detached(this);
+            }
+        }
+
         for (auto& o : observers_)
             o->detached(this);
-    };
+    }
 
     bool attach(Observer<T>* o) {
         std::lock_guard<std::mutex> lk(mutex_);
@@ -59,6 +73,23 @@ public:
         return false;
     }
 
+    void attachPriorityObserver(std::shared_ptr<Observer<T>> o) {
+        std::lock_guard<std::mutex> lk(mutex_);
+        priority_observers_.push_back(o);
+        o->attached(this);
+    }
+
+    void detachPriorityObserver(Observer<T>* o){
+        std::lock_guard<std::mutex> lk(mutex_);
+        for(auto it=priority_observers_.begin(); it != priority_observers_.end(); ++it){
+            if(auto so = it->lock()){
+                if(so.get() == o) {
+                    priority_observers_.erase(it);
+                }
+            }
+        }
+    }
+
     bool detach(Observer<T>* o) {
         std::lock_guard<std::mutex> lk(mutex_);
         if (o and observers_.erase(o)) {
@@ -76,27 +107,53 @@ public:
 protected:
     void notify(T data) {
         std::lock_guard<std::mutex> lk(mutex_);
-        for (auto observer : observers_)
+        for(auto it=priority_observers_.begin(); it != priority_observers_.end();) {
+            if(auto so = it->lock()) {
+                try {
+                    so->update(this,data);
+                    ++it;
+                } catch (std::exception& e) {
+                    JAMI_ERR() << e.what();
+                }
+            } else {
+                it = priority_observers_.erase(it);
+
+            }
+        }
+
+        for (auto observer : observers_) {
             observer->update(this, data);
+        }
     }
 
 private:
     NON_COPYABLE(Observable<T>);
 
+protected:
     std::mutex mutex_; // lock observers_
+    std::list<std::weak_ptr<Observer<T>>> priority_observers_;
     std::set<Observer<T>*> observers_;
 };
 
+
+template <typename T>
+class PublishObservable : public Observable<T> {
+public:
+    void publish(T data) {
+        this->notify(data);
+    }
+};
+
 /*=== Observer =============================================================*/
 
 template <typename T>
 class Observer
 {
 public:
-    virtual ~Observer() {};
+    virtual ~Observer() {}
     virtual void update(Observable<T>*, const T&) = 0;
-    virtual void attached(Observable<T>*) {};
-    virtual void detached(Observable<T>*) {};
+    virtual void attached(Observable<T>*) {}
+    virtual void detached(Observable<T>*) {}
 };
 
 
@@ -105,11 +162,64 @@ class FuncObserver : public Observer<T>
 {
 public:
     using F = std::function<void(const T&)>;
-    FuncObserver(F f) : f_(f) {};
-    virtual ~FuncObserver() {};
+    FuncObserver(F f) : f_(f) {}
+    virtual ~FuncObserver() {}
     void update(Observable<T>*, const T& t) override { f_(t); }
 private:
     F f_;
 };
 
+/*=== PublishMapSubject ====================================================*/
+
+template <typename T1, typename T2>
+class PublishMapSubject : public Observer<T1> , public Observable<T2> {
+public:
+    using F = std::function<T2(const T1&)>;
+
+    PublishMapSubject(F f) : map_{f} {}
+
+    void update(Observable<T1>*, const T1& t) override {
+        this->notify(map_(t));
+    }
+
+    /**
+     * @brief attached
+     * Here we just make sure that the PublishMapSubject is only attached to one
+     * Observable at a time.
+     * @param srcObs
+     */
+    virtual void attached(Observable<T1>* srcObs) override {
+        if(obs_!=nullptr && obs_ != srcObs) {
+            obs_->detach(this);
+            obs_ = srcObs;
+        }
+    }
+
+    /**
+     * @brief detached
+     * Since a MapSubject is only attached to one Observable, when detached
+     * We should detach all of it observers
+     */
+    virtual void detached(Observable<T1>*) override {
+        std::lock_guard<std::mutex> lk(this->mutex_);
+        JAMI_WARN() << "PublishMapSubject: detaching observers";
+        for (auto& o : this->observers_)
+            o->detached(this);
+    }
+
+    /**
+     * @brief ~PublishMapSubject()
+     * Detach all observers to avoid making them call this observable when
+     * destroyed
+    **/
+    ~PublishMapSubject() {
+        JAMI_WARN() << "~PublishMapSubject()";
+        detached(nullptr);
+    }
+
+private:
+    F map_;
+    Observable<T1>* obs_ = nullptr;
+};
+
 }; // namespace jami

src/plugin/callservicesmanager.h

@@ -0,0 +1,217 @@
+/*
+ *  Copyright (C) 2004-2019 Savoir-faire Linux Inc.
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+ */
+
+#pragma once
+// Utils
+#include "noncopyable.h"
+// Plugin Manager
+#include "pluginmanager.h"
+#include "streamdata.h"
+#include "mediahandler.h"
+// STL
+#include <list>
+
+namespace jami {
+using MediaHandlerPtr = std::unique_ptr<MediaHandler>;
+using CallMediaHandlerPtr = std::unique_ptr<CallMediaHandler>;
+using AVSubjectSPtr = std::weak_ptr<Observable<AVFrame*>>;
+
+class CallServicesManager{
+
+public:
+    CallServicesManager(PluginManager& pm) {
+        registerComponentsLifeCycleManagers(pm);
+    }
+
+    /**
+    *   unload all media handlers
+    **/
+    ~CallServicesManager(){
+        callMediaHandlers.clear();
+    }
+
+    NON_COPYABLE(CallServicesManager);
+
+public:
+    /**
+     * @brief notifyAllAVSubject
+     * @param subject
+     * @param av
+     * @param local
+     * @param peerId
+     * This function is called whenever there is a new AVFrame subject available
+     */
+    void notifyAllAVSubject(const StreamData& data, AVSubjectSPtr& subject) {
+        for(auto& pair : callMediaHandlers) {
+            auto& callMediaHandlerPtr = pair.second;
+            if(pair.first) {
+                notifyAVSubject(callMediaHandlerPtr, data, subject);
+            }
+        }
+    }
+
+    /**
+     * @brief createAVSubject
+     * @param data
+     * Creates an av frame subject with properties StreamData
+     */
+    void createAVSubject(const StreamData& data, AVSubjectSPtr subject){
+        // This guarantees unicity of subjects by id
+        callAVsubjects.push_back(std::make_pair(data, subject));
+        auto inserted = callAVsubjects.back();
+        notifyAllAVSubject(inserted.first, inserted.second);
+    }
+
+    /**
+     * @brief registerComponentsLifeCycleManagers
+     * Exposes components life cycle managers to the main API
+     */
+    void registerComponentsLifeCycleManagers(PluginManager& pm) {
+
+        auto registerCallMediaHandler = [this](void* data) {
+            CallMediaHandlerPtr ptr{(static_cast<CallMediaHandler*>(data))};
+
+            if(ptr) {
+                callMediaHandlers.push_back(std::make_pair(false, std::move(ptr)));
+            }
+            return 0;
+        };
+
+        auto unregisterMediaHandler = [this](void* data) {
+            for(auto it = callMediaHandlers.begin(); it != callMediaHandlers.end(); ++it) {
+                if(it->second.get() == data) {
+                    callMediaHandlers.erase(it);
+                }
+            }
+            return 0;
+        };
+
+        pm.registerComponentManager("CallMediaHandlerManager",
+                                    registerCallMediaHandler, unregisterMediaHandler);
+    }
+
+    /**
+     * @brief listCallMediaHandlers
+     * List all call media handlers
+     * @return
+     */
+    std::vector<std::string> listCallMediaHandlers() {
+        std::vector<std::string> res;
+        for(const auto& pair : callMediaHandlers) {
+            if(pair.second) {
+              res.push_back(getCallHandlerId(pair.second));
+            }
+        }
+        return res;
+    }
+
+    /**
+     * @brief toggleCallMediaHandler
+     * Toggle CallMediaHandler, if on, notify with new subjects
+     * if off, detach it
+     * @param id
+     */
+    void toggleCallMediaHandler(const std::string& id, const bool toggle) {
+        for(auto& pair : callMediaHandlers) {
+            if(pair.second && getCallHandlerId(pair.second) == id) {
+                pair.first = toggle;
+                if(pair.first) {
+                    listAvailableSubjects(pair.second);
+                } else {
+                    pair.second->detach();
+                }
+            }
+        }
+    }
+
+    /**
+     * @brief getCallMediaHandlerDetails
+     * @param id of the call media handler
+     * @return map of Call Media Handler Details
+     */
+    std::map<std::string, std::string >
+    getCallMediaHandlerDetails(const std::string& id) {
+        for(auto& pair : callMediaHandlers) {
+            if(pair.second && getCallHandlerId(pair.second) == id) {
+                return pair.second->getCallMediaHandlerDetails();
+            }
+        }
+        return {};
+    }
+
+private:
+
+    /**
+     * @brief notifyAVSubject
+     * @param callMediaHandlerPtr
+     * @param data
+     * @param subject
+     */
+    void notifyAVSubject(CallMediaHandlerPtr& callMediaHandlerPtr,
+                              const StreamData& data,
+                         AVSubjectSPtr& subject) {
+        if(auto soSubject = subject.lock()) {
+            callMediaHandlerPtr->notifyAVFrameSubject(data, soSubject);
+        }
+    }
+
+    /**
+     * @brief listAvailableSubjects
+     * @param callMediaHandlerPtr
+     * This functions lets the call media handler component know which subjects are available
+     */
+    void listAvailableSubjects(CallMediaHandlerPtr& callMediaHandlerPtr) {
+        for(auto it=callAVsubjects.begin(); it != callAVsubjects.end(); ++it) {
+            notifyAVSubject(callMediaHandlerPtr, it->first, it->second);
+        }
+    }
+
+    /**
+     * @brief getCallHandlerId
+     * Returns the callMediaHandler id from a callMediaHandler pointer
+     * @param callMediaHandler
+     * @return string id
+     */
+    std::string getCallHandlerId(const CallMediaHandlerPtr& callMediaHandler) {
+        if(callMediaHandler) {
+            std::ostringstream callHandlerIdStream;
+            callHandlerIdStream << callMediaHandler.get();
+            return callHandlerIdStream.str();
+        }
+        return "";
+    }
+
+private:
+    /**
+     * @brief callMediaHandlers
+     * Components that a plugin can register through registerCallMediaHandler service
+     * These objects can then be notified with notify notifyAVFrameSubject
+     * whenever there is a new CallAVSubject like a video receive
+     */
+    std::list<std::pair<bool, CallMediaHandlerPtr>> callMediaHandlers;
+    /**
+     * @brief callAVsubjects
+     * When there is a SIPCall, CallAVSubjects are created there
+     * Here we keep a reference to them in order to make them interact with
+     * CallMediaHandlers
+     * It is pushed to this list list
+     */
+    std::list<std::pair<const StreamData, AVSubjectSPtr>> callAVsubjects;
+};
+
+}

src/plugin/jamipluginmanager.h

@@ -22,6 +22,9 @@
 #include "archiver.h"
 #include "pluginmanager.h"
 
+//Services
+#include "callservicesmanager.h"
+
 #include <vector>
 #include <map>
 #include <string>
@@ -31,7 +34,7 @@ namespace jami {
 class JamiPluginManager
 {
 public:
-    JamiPluginManager() {
+    JamiPluginManager() : csm_{pm_}{
         registerServices();
     }
     // TODO : improve getPluginDetails
@@ -117,6 +120,12 @@ public:
 
     bool resetPluginPreferencesValuesMap(const std::string& rootPath);
 
+public:
+
+    CallServicesManager& getCallServicesManager() {
+        return csm_;
+    }
+
 private:
 
     NON_COPYABLE(JamiPluginManager);
@@ -189,6 +198,10 @@ private:
 private:
     PluginManager pm_;
     std::map<std::string, std::map<std::string, std::string>> pluginDetailsMap_;
+
+//Services
+private:
+    CallServicesManager csm_;
 };
 }
 

src/plugin/mediahandler.h

@@ -0,0 +1,69 @@
+/*
+ *  Copyright (C) 2004-2019 Savoir-faire Linux Inc.
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+ */
+#pragma once
+#include "streamdata.h"
+#include "observer.h"
+#include <libavutil/frame.h>
+#include <string>
+#include <memory>
+#include <map>
+
+namespace jami {
+
+using avSubjectPtr = std::shared_ptr<Observable<AVFrame*>>;
+
+/**
+ * @brief The MediaHandler class
+ * Is the main object of the plugin
+ */
+class MediaHandler{
+
+public:
+    virtual ~MediaHandler() = default;
+    /**
+     * @brief id
+     * The id is the path of the plugin that created this MediaHandler
+     * @return
+     */
+    std::string id() const { return id_;}
+    virtual void setId(const std::string& id) final {id_ = id;}
+    /**
+     * @brief setPreferenceAttribute
+     * Sets a preference attribute to the new value
+     * @param key
+     * @param value
+     */
+    virtual void setPreferenceAttribute(const std::string& key, const std::string& value) {
+        (void)key;(void)value;
+    }
+
+private:
+    std::string id_;
+};
+
+/**
+ * @brief The CallMediaHandler class
+ * It can hold multiple streams of data, and do processing on them
+ */
+class CallMediaHandler: public MediaHandler {
+public:
+    virtual void notifyAVFrameSubject(const StreamData& data, avSubjectPtr subject) = 0;
+    virtual std::map<std::string, std::string> getCallMediaHandlerDetails() = 0;
+    virtual void detach() = 0;
+};
+}

src/plugin/streamdata.h

@@ -0,0 +1,30 @@
+/*
+ *  Copyright (C) 2004-2019 Savoir-faire Linux Inc.
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+ */
+#pragma once
+#include <string>
+
+enum class StreamType { audio, video };
+
+struct StreamData {
+    StreamData(const std::string& i, bool d, StreamType&& t, const std::string& s) :
+        id{std::move(i)}, direction{d}, type{t}, source{std::move(s)} {}
+    const std::string id;
+    const bool direction;
+    const StreamType type;
+    const std::string source;
+};

src/sip/sipcall.cpp

@@ -41,12 +41,15 @@
 #include "dring/media_const.h"
 #include "client/ring_signal.h"
 #include "ice_transport.h"
+//Plugin manager
+#include "plugin/jamipluginmanager.h"
 
 #ifdef ENABLE_VIDEO
 #include "client/videomanager.h"
 #include "video/video_rtp_session.h"
 #include "dring/videomanager_interface.h"
 #include <chrono>
+#include <libavutil/display.h>
 #endif
 
 #include "errno.h"
@@ -115,6 +118,44 @@ SIPCall::getSIPAccount() const
     return static_cast<SIPAccountBase&>(getAccount());
 }
 
+void SIPCall::createCallAVStreams()
+{
+    if(hasVideo()){
+        /**
+        *   Map: maps the VideoFrame to an AVFrame
+        **/
+        auto map = [](const std::shared_ptr<jami::MediaFrame> m)->AVFrame* {
+            return std::static_pointer_cast<VideoFrame>(m)->pointer();
+        };
+
+        // Preview
+        auto& videoPreview = videortp_->getVideoLocal();
+
+        if(videoPreview) {
+            auto previewSubject = std::make_shared<MediaStreamSubject>(map);
+            StreamData previewStreamData{getCallId(), 0, StreamType::video, getPeerNumber()};
+            createCallAVStream(previewStreamData, *videoPreview, previewSubject);
+        }
+
+        // Receive
+        auto& videoReceive = videortp_->getVideoReceive();
+
+        if(videoReceive) {
+            auto receiveSubject = std::make_shared<MediaStreamSubject>(map);
+            StreamData receiveStreamData{getCallId(), 1, StreamType::video, getPeerNumber()};
+            createCallAVStream(receiveStreamData, *videoReceive, receiveSubject);
+        }
+    }
+}
+
+void SIPCall::createCallAVStream(const StreamData& StreamData, MediaStream& streamSource,
+                                 const std::shared_ptr<MediaStreamSubject>& mediaStreamSubject){
+    callAVStreams.push_back(mediaStreamSubject);
+    auto& inserted = callAVStreams.back();
+    streamSource.attachPriorityObserver(inserted);
+    jami::Manager::instance().getJamiPluginManager().getCallServicesManager().createAVSubject(StreamData, inserted);
+}
+
 void
 SIPCall::setCallMediaLocal()
 {
@@ -1073,6 +1114,9 @@ SIPCall::startAllMedia()
         }
         remainingRequest_ = Request::NoRequest;
     }
+
+    // Create AVStreams associated with the call
+    createCallAVStreams();
 }
 
 void

src/sip/sipcall.h

@@ -33,8 +33,10 @@
 #include "sip_utils.h"
 
 #ifdef ENABLE_VIDEO
+#include "media/video/video_receive_thread.h"
 #include "media/video/video_rtp_session.h"
 #endif
+#include "plugin/streamdata.h"
 
 #include "noncopyable.h"
 
@@ -247,9 +249,35 @@ private:
 
     NON_COPYABLE(SIPCall);
 
+
     IceTransport* getIceMediaTransport() const {
         return tmpMediaTransport_ ? tmpMediaTransport_.get() : mediaTransport_.get();
     }
+
+    /**
+     * Call Streams and some typedefs
+     */
+    using MediaStream = Observable<std::shared_ptr<MediaFrame>>;
+    using MediaStreamSubject = PublishMapSubject<std::shared_ptr<MediaFrame>, AVFrame*>;
+
+    /**
+     * @brief createCallAVStream
+     * Creates a call AV stream like video input, video receive, audio input or audio receive
+     * @param StreamData The type of the stream (audio/video, input/output,
+     * @param streamSource
+     * @param mediaStreamSubject
+     */
+    void createCallAVStream(const StreamData& StreamData, MediaStream& streamSource,
+                            const std::shared_ptr<MediaStreamSubject>& mediaStreamSubject);
+    /**
+     * @brief createCallAVStreams
+     * Creates all Call AV Streams (2 if audio, 4 if audio video)
+     */
+    void createCallAVStreams();
+
+    std::list<std::shared_ptr<MediaStreamSubject>> callAVStreams;
+
+
     void setCallMediaLocal();
 
     void waitForIceAndStartMedia();