git.sesse.net Git - casparcg/blob - core/monitor/monitor.h

   1 #pragma once
   2
   3 #include <common/reactive.h>
   4
   5 #include <functional>
   6 #include <memory>
   7 #include <ostream>
   8 #include <string>
   9 #include <type_traits>
  10 #include <vector>
  11
  12 #include <boost/variant.hpp>
  13 #include <boost/lexical_cast.hpp>
  14 #include <boost/chrono.hpp>
  15
  16 #include <tbb/cache_aligned_allocator.h>
  17
  18 namespace boost {
  19 namespace detail { namespace variant {
  20
  21 template <>
  22 struct has_nothrow_move<std::string>
  23 : mpl::true_
  24 {
  25 };
  26
  27 template <>
  28 struct has_nothrow_move<std::vector<int8_t>>
  29 : mpl::true_
  30 {
  31 };
  32
  33 template <>
  34 struct has_nothrow_move<boost::chrono::duration<double, boost::ratio<1, 1>>>
  35 : mpl::true_
  36 {
  37 };
  38
  39 }}}
  40
  41 namespace caspar { namespace monitor {
  42
  43 // path
  44
  45 class path sealed
  46 {
  47 public:
  48
  49         // Static Members
  50
  51         // Constructors
  52
  53         path();
  54         path(const char* path);
  55         path(std::string path);
  56
  57         path(const path& other);
  58         path(path&& other);
  59
  60         // Methods
  61
  62         path& operator=(path other);
  63         path& operator%=(path other);
  64
  65         template<typename T>
  66         typename std::enable_if<!std::is_same<typename std::decay<T>::type, path>::value, path&>::type operator%=(T&& value)
  67         {
  68                 auto str = boost::lexical_cast<std::string>(std::forward<T>(value));
  69
  70                 if(!str.empty())
  71                         str_ += (str.front() != '/' ? "/" : "") + std::move(str);
  72
  73                 return *this;
  74         }
  75
  76         path& operator%=(const char* value)
  77         {
  78                 return *this %= std::string(value);
  79         }
  80
  81         void swap(path& other);
  82
  83         // Properties
  84
  85         const std::string& str() const;
  86         bool empty() const;
  87 private:
  88         std::string str_;
  89 };
  90
  91 template<typename T>
  92 path operator%(path path, T&& value)
  93 {
  94         return std::move(path %= std::forward<T>(value));
  95 }
  96
  97 std::ostream& operator<<(std::ostream& o, const path& p);
  98
  99 // param
 100
 101 typedef boost::chrono::duration<double, boost::ratio<1, 1>> duration;
 102
 103 typedef boost::variant<bool, int32_t, int64_t, float, double, std::string, std::wstring, std::vector<int8_t>, duration> param;
 104
 105 std::ostream& operator<<(std::ostream& o, const param& p);
 106
 107 // event
 108
 109 class event sealed
 110 {
 111 public:
 112
 113         // Static Members
 114
 115         typedef std::vector<param, tbb::cache_aligned_allocator<param>> params_t;
 116
 117         // Constructors
 118
 119         event(path path);
 120         event(path path, params_t params);
 121         event(const event& other);
 122         event(event&& other);
 123
 124         // Methods
 125
 126         event& operator=(event other);
 127
 128         void swap(event& other);
 129
 130         template<typename T>
 131         event& operator%(T&& value)
 132         {
 133                 params_.push_back(std::forward<T>(value));
 134                 return *this;
 135         }
 136
 137         event propagate(path path) const;
 138
 139         // Properties
 140
 141         const path&             path() const;
 142         const params_t& params() const;
 143 private:
 144         monitor::path   path_;
 145         params_t                params_;
 146 };
 147
 148 std::ostream& operator<<(std::ostream& o, const event& e);
 149
 150 // reactive
 151
 152 typedef reactive::observable<monitor::event>    observable;
 153 typedef reactive::observer<monitor::event>              observer;
 154 typedef reactive::subject<monitor::event>               subject;
 155
 156 class basic_subject sealed : public reactive::subject<monitor::event>
 157 {
 158         basic_subject(const basic_subject&);
 159         basic_subject& operator=(const basic_subject&);
 160
 161         class impl : public observer
 162         {
 163         public:
 164                 impl(monitor::path path = monitor::path())
 165                         : impl_()
 166                         , path_(std::move(path))
 167                 {
 168                 }
 169
 170                 impl(impl&& other)
 171                         : impl_(std::move(other.impl_))
 172                         , path_(std::move(other.path_))
 173                 {
 174                 }
 175
 176                 impl& operator=(impl&& other)
 177                 {
 178                         impl_ = std::move(other.impl_);
 179                         path_ = std::move(other.path_);
 180                 }
 181
 182                 void on_next(const monitor::event& e) override
 183                 {
 184                         impl_.on_next(path_.empty() ? e : e.propagate(path_));
 185                 }
 186
 187                 void subscribe(const observer_ptr& o)
 188                 {
 189                         impl_.subscribe(o);
 190                 }
 191
 192                 void unsubscribe(const observer_ptr& o)
 193                 {
 194                         impl_.unsubscribe(o);
 195                 }
 196
 197         private:
 198                 reactive::basic_subject_impl<monitor::event>    impl_;
 199                 monitor::path                                                                   path_;
 200         };
 201
 202 public:
 203
 204         // Static Members
 205
 206         // Constructors
 207
 208         basic_subject(monitor::path path = monitor::path())
 209                 : impl_(std::make_shared<impl>(std::move(path)))
 210
 211         {
 212         }
 213
 214         basic_subject(basic_subject&& other)
 215                 : impl_(std::move(other.impl_))
 216         {
 217         }
 218
 219         // Methods
 220
 221         basic_subject& operator=(basic_subject&& other)
 222         {
 223                 impl_ = std::move(other.impl_);
 224         }
 225
 226         operator std::weak_ptr<observer>()
 227         {
 228                 return impl_;
 229         }
 230
 231         // observable
 232
 233         void subscribe(const observer_ptr& o) override
 234         {
 235                 impl_->subscribe(o);
 236         }
 237
 238         void unsubscribe(const observer_ptr& o) override
 239         {
 240                 impl_->unsubscribe(o);
 241         }
 242
 243         // observer
 244
 245         void on_next(const monitor::event& e) override
 246         {
 247                 impl_->on_next(e);
 248         }
 249
 250         // Properties
 251
 252 private:
 253         std::shared_ptr<impl> impl_;
 254 };
 255
 256 inline subject& operator<<(subject& s, const event& e)
 257 {
 258         s.on_next(e);
 259         return s;
 260 }
 261
 262 }}