Add metadata methods in pipeline (#700)

Author: starwarfan

source/core/owt_base/InternalIn.cpp

@@ -41,12 +41,18 @@ void InternalIn::onFeedback(const FeedbackMsg& msg)
 void InternalIn::onTransportData(char* buf, int len)
 {
     Frame* frame = nullptr;
+    MetaData* metadata = nullptr;
     switch (buf[0]) {
         case TDT_MEDIA_FRAME:
             frame = reinterpret_cast<Frame*>(buf + 1);
             frame->payload = reinterpret_cast<uint8_t*>(buf + 1 + sizeof(Frame));
             deliverFrame(*frame);
             break;
+        case TDT_MEDIA_METADATA:
+            metadata = reinterpret_cast<MetaData*>(buf + 1);
+            metadata->payload = reinterpret_cast<uint8_t*>(buf + 1 + sizeof(MetaData));
+            deliverMetaData(*metadata);
+            break;
         default:
             break;
     }

source/core/owt_base/InternalOut.cpp

@@ -31,6 +31,16 @@ void InternalOut::onFrame(const Frame& frame)
     m_transport->sendData(sendBuffer, header_len + 1, reinterpret_cast<char*>(const_cast<uint8_t*>(frame.payload)), frame.length);
 }
 
+void InternalOut::onMetaData(const MetaData& metadata)
+{
+    char sendBuffer[sizeof(MetaData) + 1];
+    size_t header_len = sizeof(MetaData);
+
+    sendBuffer[0] = TDT_MEDIA_METADATA;
+    memcpy(&sendBuffer[1], reinterpret_cast<char*>(const_cast<MetaData*>(&metadata)), header_len);
+    m_transport->sendData(sendBuffer, header_len + 1, reinterpret_cast<char*>(const_cast<uint8_t*>(metadata.payload)), metadata.length);
+}
+
 void InternalOut::onTransportData(char* buf, int len)
 {
     switch (buf[0]) {

source/core/owt_base/InternalOut.h

@@ -16,6 +16,8 @@ class InternalOut : public FrameDestination, public RawTransportListener {
     virtual ~InternalOut();
 
     void onFrame(const Frame&);
+    void onMetaData(const MetaData&);
+
 
     void onTransportData(char*, int len);
     void onTransportError() { }

source/core/owt_base/MediaFramePipeline.cpp

@@ -99,6 +99,22 @@ void FrameSource::deliverFrame(const Frame& frame)
     }
 }
 
+void FrameSource::deliverMetaData(const MetaData& metadata)
+{
+    {
+        boost::shared_lock<boost::shared_mutex> lock(m_audio_dests_mutex);
+        for (auto it = m_audio_dests.begin(); it != m_audio_dests.end(); ++it) {
+            (*it)->onMetaData(metadata);
+        }
+    }
+    {
+        boost::shared_lock<boost::shared_mutex> lock(m_video_dests_mutex);
+        for (auto it = m_video_dests.begin(); it != m_video_dests.end(); ++it) {
+            (*it)->onMetaData(metadata);
+        }
+    }
+}
+
 //=========================================================================================
 
 void FrameDestination::setAudioSource(FrameSource* src)

source/core/owt_base/MediaFramePipeline.h

@@ -85,6 +85,16 @@ struct Frame {
     MediaSpecInfo   additionalInfo;
 };
 
+enum MetaDataType {
+    META_DATA_OWNER_ID = 0,
+};
+
+struct MetaData {
+    MetaDataType type;
+    uint8_t* payload;
+    uint32_t length;
+};
+
 inline FrameFormat getFormat(const std::string& codec) {
     if (codec == "vp8") {
         return owt_base::FRAME_FORMAT_VP8;
@@ -210,6 +220,7 @@ enum FeedbackType {
 enum FeedbackCmd {
     REQUEST_KEY_FRAME,
     SET_BITRATE,
+    REQUEST_OWNER_ID,
     RTCP_PACKET  // FIXME: Temporarily use FeedbackMsg to carry audio rtcp-packets due to the premature AudioFrameConstructor implementation.
 };
 
@@ -245,6 +256,7 @@ class FrameSource {
 
 protected:
     void deliverFrame(const Frame&);
+    void deliverMetaData(const MetaData&);
 
 private:
     std::list<FrameDestination*> m_audio_dests;
@@ -262,6 +274,7 @@ class FrameDestination {
     virtual ~FrameDestination() { }
 
     virtual void onFrame(const Frame&) = 0;
+    virtual void onMetaData(const MetaData&) {}
     virtual void onVideoSourceChanged() {}
 
     void setAudioSource(FrameSource*);

source/core/owt_base/RawTransport.h

@@ -18,6 +18,7 @@ namespace owt_base {
 
 const char TDT_FEEDBACK_MSG = 0x5A;
 const char TDT_MEDIA_FRAME = 0x8F;
+const char TDT_MEDIA_METADATA = 0x3A;
 
 enum Protocol {
     TCP = 0,