]> git.sesse.net Git - casparcg/blob - core/consumer/port.cpp
Merged asynchronous invocation of consumers from 2.0
[casparcg] / core / consumer / port.cpp
1 #include "../StdAfx.h"
2
3 #include "port.h"
4
5 #include "frame_consumer.h"
6 #include "../frame/frame.h"
7
8 namespace caspar { namespace core {
9
10 struct port::impl
11 {
12         monitor::basic_subject                          event_subject_;
13         std::shared_ptr<frame_consumer>         consumer_;
14         int                                                                     index_;
15         int                                                                     channel_index_;
16 public:
17         impl(int index, int channel_index, spl::shared_ptr<frame_consumer> consumer)
18                 : event_subject_(monitor::path("port") % index)
19                 , consumer_(std::move(consumer))
20                 , index_(index)
21                 , channel_index_(channel_index)
22         {
23                 consumer_->subscribe(event_subject_);
24         }
25         
26         void video_format_desc(const struct video_format_desc& format_desc)
27         {
28                 consumer_->initialize(format_desc, channel_index_);
29         }
30                 
31         boost::unique_future<bool> send(const_frame frame)
32         {
33                 event_subject_ << monitor::event("type") % consumer_->name();
34                 return consumer_->send(std::move(frame));
35         }
36         
37         int index() const
38         {
39                 return index_;
40         }
41
42         int buffer_depth() const
43         {
44                 return consumer_->buffer_depth();
45         }
46
47         bool has_synchronization_clock() const
48         {
49                 return consumer_->has_synchronization_clock();
50         }
51
52         boost::property_tree::wptree info() const
53         {
54                 return consumer_->info();
55         }
56 };
57
58 port::port(int index, int channel_index, spl::shared_ptr<frame_consumer> consumer) : impl_(new impl(index, channel_index, std::move(consumer))){}
59 port::port(port&& other) : impl_(std::move(other.impl_)){}
60 port::~port(){}
61 port& port::operator=(port&& other){impl_ = std::move(other.impl_); return *this;}
62 boost::unique_future<bool> port::send(const_frame frame){return impl_->send(std::move(frame));} 
63 void port::subscribe(const monitor::observable::observer_ptr& o){impl_->event_subject_.subscribe(o);}
64 void port::unsubscribe(const monitor::observable::observer_ptr& o){impl_->event_subject_.unsubscribe(o);}
65 void port::video_format_desc(const struct video_format_desc& format_desc){impl_->video_format_desc(format_desc);}
66 int port::buffer_depth() const{return impl_->buffer_depth();}
67 bool port::has_synchronization_clock() const{return impl_->has_synchronization_clock();}
68 boost::property_tree::wptree port::info() const{return impl_->info();}
69 }}