Implement listener interfaces.

src/Notifier.cpp

@@ -0,0 +1,187 @@
+/*----------------------------------------------------------------------------*/
+/* Copyright (c) FIRST 2015-2016. All Rights Reserved.                        */
+/* Open Source Software - may be modified and shared by FRC teams. The code   */
+/* must be accompanied by the FIRST BSD license file in the root directory of */
+/* the project.                                                               */
+/*----------------------------------------------------------------------------*/
+
+#include "Notifier.h"
+
+#include <queue>
+#include <vector>
+
+#include "Handle.h"
+
+using namespace cs;
+
+ATOMIC_STATIC_INIT(Notifier)
+bool Notifier::s_destroyed = false;
+
+namespace {
+// Vector which provides an integrated freelist for removal and reuse of
+// individual elements.
+template <typename T>
+class UidVector {
+ public:
+  typedef typename std::vector<T>::size_type size_type;
+
+  size_type size() const { return m_vector.size(); }
+  T& operator[](size_type i) { return m_vector[i]; }
+  const T& operator[](size_type i) const { return m_vector[i]; }
+
+  // Add a new T to the vector.  If there are elements on the freelist,
+  // reuses the last one; otherwise adds to the end of the vector.
+  // Returns the resulting element index (+1).
+  template <class... Args>
+  unsigned int emplace_back(Args&&... args) {
+    unsigned int uid;
+    if (m_free.empty()) {
+      uid = m_vector.size();
+      m_vector.emplace_back(std::forward<Args>(args)...);
+    } else {
+      uid = m_free.back();
+      m_free.pop_back();
+      m_vector[uid] = T(std::forward<Args>(args)...);
+    }
+    return uid + 1;
+  }
+
+  // Removes the identified element by replacing it with a default-constructed
+  // one.  The element is added to the freelist for later reuse.
+  void erase(unsigned int uid) {
+    --uid;
+    if (uid >= m_vector.size() || !m_vector[uid]) return;
+    m_free.push_back(uid);
+    m_vector[uid] = T();
+  }
+
+ private:
+  std::vector<T> m_vector;
+  std::vector<unsigned int> m_free;
+};
+
+}  // anonymous namespace
+
+class Notifier::Thread : public wpi::SafeThread {
+ public:
+  Thread(std::function<void()> on_start, std::function<void()> on_exit)
+      : m_on_start(on_start), m_on_exit(on_exit) {}
+
+  void Main();
+
+  struct Listener {
+    Listener() = default;
+    Listener(std::function<void(const RawEvent& event)> callback_,
+             int eventMask_)
+        : callback(callback_), eventMask(eventMask_) {}
+
+    explicit operator bool() const { return bool(callback); }
+
+    std::string prefix;
+    std::function<void(const RawEvent& event)> callback;
+    int eventMask;
+  };
+  UidVector<Listener> m_listeners;
+
+  std::queue<RawEvent> m_notifications;
+
+  std::function<void()> m_on_start;
+  std::function<void()> m_on_exit;
+};
+
+Notifier::Notifier() { s_destroyed = false; }
+
+Notifier::~Notifier() { s_destroyed = true; }
+
+void Notifier::Start() {
+  auto thr = m_owner.GetThread();
+  if (!thr) m_owner.Start(new Thread(m_on_start, m_on_exit));
+}
+
+void Notifier::Stop() { m_owner.Stop(); }
+
+void Notifier::Thread::Main() {
+  if (m_on_start) m_on_start();
+
+  std::unique_lock<std::mutex> lock(m_mutex);
+  while (m_active) {
+    while (m_notifications.empty()) {
+      m_cond.wait(lock);
+      if (!m_active) goto done;
+    }
+
+    while (!m_notifications.empty()) {
+      if (!m_active) goto done;
+      auto item = std::move(m_notifications.front());
+      m_notifications.pop();
+
+      // Use index because iterator might get invalidated.
+      for (std::size_t i = 0; i < m_listeners.size(); ++i) {
+        if (!m_listeners[i]) continue;  // removed
+
+        // Event type must be within requested set for this listener.
+        if ((item.type & m_listeners[i].eventMask) == 0) continue;
+
+        // make a copy of the callback so we can safely release the mutex
+        auto callback = m_listeners[i].callback;
+
+        // Don't hold mutex during callback execution!
+        lock.unlock();
+        callback(item);
+        lock.lock();
+      }
+    }
+  }
+
+done:
+  if (m_on_exit) m_on_exit();
+}
+
+int Notifier::AddListener(
+    std::function<void(const RawEvent& event)> callback, int eventMask) {
+  Start();
+  auto thr = m_owner.GetThread();
+  return thr->m_listeners.emplace_back(callback, eventMask);
+}
+
+void Notifier::RemoveListener(int uid) {
+  auto thr = m_owner.GetThread();
+  if (!thr) return;
+  thr->m_listeners.erase(uid);
+}
+
+void Notifier::NotifySource(llvm::StringRef name, CS_Source source,
+                            RawEvent::Type type) {
+  auto thr = m_owner.GetThread();
+  if (!thr) return;
+  thr->m_notifications.emplace(name, source, type);
+  thr->m_cond.notify_one();
+}
+
+void Notifier::NotifySourceVideoMode(llvm::StringRef name, CS_Source source,
+                                     const VideoMode& mode) {
+  auto thr = m_owner.GetThread();
+  if (!thr) return;
+  thr->m_notifications.emplace(name, source, mode);
+  thr->m_cond.notify_one();
+}
+
+void Notifier::NotifySourceProperty(llvm::StringRef name, CS_Source source,
+                                    RawEvent::Type type, int property,
+                                    CS_PropertyType propertyType, int value,
+                                    llvm::StringRef valueStr) {
+  auto thr = m_owner.GetThread();
+  if (!thr) return;
+  thr->m_notifications.emplace(name, source, type,
+                               Handle{source, property, Handle::kProperty},
+                               propertyType, value, valueStr);
+  thr->m_cond.notify_one();
+}
+
+void Notifier::NotifySink(llvm::StringRef name, CS_Sink sink,
+                          RawEvent::Type type) {
+  auto thr = m_owner.GetThread();
+  if (!thr) return;
+  thr->m_notifications.emplace(name, sink, type);
+  thr->m_cond.notify_one();
+}

src/Notifier.h

@@ -0,0 +1,67 @@
+/*----------------------------------------------------------------------------*/
+/* Copyright (c) FIRST 2015-2016. All Rights Reserved.                        */
+/* Open Source Software - may be modified and shared by FRC teams. The code   */
+/* must be accompanied by the FIRST BSD license file in the root directory of */
+/* the project.                                                               */
+/*----------------------------------------------------------------------------*/
+
+#ifndef CAMERASERVER_NOTIFIER_H_
+#define CAMERASERVER_NOTIFIER_H_
+
+#include <functional>
+
+#include "support/atomic_static.h"
+#include "support/SafeThread.h"
+#include "cameraserver_cpp.h"
+
+namespace cs {
+
+class Notifier {
+  friend class NotifierTest;
+
+ public:
+  static Notifier& GetInstance() {
+    ATOMIC_STATIC(Notifier, instance);
+    return instance;
+  }
+  ~Notifier();
+
+  void Start();
+  void Stop();
+
+  static bool destroyed() { return s_destroyed; }
+
+  void SetOnStart(std::function<void()> on_start) { m_on_start = on_start; }
+  void SetOnExit(std::function<void()> on_exit) { m_on_exit = on_exit; }
+
+  int AddListener(std::function<void(const RawEvent& event)> callback,
+                  int eventMask);
+  void RemoveListener(int uid);
+
+  // Notification events
+  void NotifySource(llvm::StringRef name, CS_Source source,
+                    RawEvent::Type type);
+  void NotifySourceVideoMode(llvm::StringRef name, CS_Source source,
+                             const VideoMode& mode);
+  void NotifySourceProperty(llvm::StringRef name, CS_Source source,
+                            RawEvent::Type type, int property,
+                            CS_PropertyType propertyType, int value,
+                            llvm::StringRef valueStr);
+  void NotifySink(llvm::StringRef name, CS_Sink sink, RawEvent::Type type);
+
+ private:
+  Notifier();
+
+  class Thread;
+  wpi::SafeThreadOwner<Thread> m_owner;
+
+  std::function<void()> m_on_start;
+  std::function<void()> m_on_exit;
+
+  ATOMIC_STATIC_DECL(Notifier)
+  static bool s_destroyed;
+};
+
+}  // namespace cs
+
+#endif  // CAMERASERVER_NOTIFIER_H_

src/cameraserver_c.cpp

@@ -211,6 +211,14 @@ void CS_ReleaseSink(CS_Sink sink, CS_Status* status) {
   return cs::ReleaseSink(sink, status);
 }
 
+void CS_SetListenerOnStart(void (*onStart)(void* data), void* data) {
+  cs::SetListenerOnStart([=]() { onStart(data); });
+}
+
+void CS_SetListenerOnExit(void (*onExit)(void* data), void* data) {
+  cs::SetListenerOnExit([=]() { onExit(data); });
+}
+
 CS_Listener CS_AddListener(void* data,
                            void (*callback)(void* data, const CS_Event* event),
                            int eventMask, int immediateNotify,
@@ -236,6 +244,8 @@ void CS_RemoveListener(CS_Listener handle, CS_Status* status) {
   return cs::RemoveListener(handle, status);
 }
 
+int CS_NotifierDestroyed(void) { return cs::NotifierDestroyed(); }
+
 CS_Source* CS_EnumerateSources(int* count, CS_Status* status) {
   llvm::SmallVector<CS_Source, 32> buf;
   auto handles = cs::EnumerateSourceHandles(buf, status);

src/cameraserver_cpp.cpp

@@ -9,6 +9,7 @@
 
 #include "llvm/SmallString.h"
 
+#include "Notifier.h"
 #include "SinkImpl.h"
 #include "SourceImpl.h"
 #include "Handle.h"
@@ -424,16 +425,35 @@ void ReleaseSink(CS_Sink sink, CS_Status* status) {
 // Listener Functions
 //
 
+void SetListenerOnStart(std::function<void()> onStart) {
+  Notifier::GetInstance().SetOnStart(onStart);
+}
+
+void SetListenerOnExit(std::function<void()> onExit) {
+  Notifier::GetInstance().SetOnExit(onExit);
+}
+
 CS_Listener AddListener(std::function<void(const RawEvent& event)> callback,
                         int eventMask, bool immediateNotify,
                         CS_Status* status) {
-  return 0;  // TODO
+  int uid = Notifier::GetInstance().AddListener(callback, eventMask);
+  if (immediateNotify) {
+    // TODO
+  }
+  return Handle{uid, Handle::kListener};
 }
 
 void RemoveListener(CS_Listener handle, CS_Status* status) {
-  // TODO
+  int uid = Handle{handle}.GetTypedIndex(Handle::kListener);
+  if (uid < 0) {
+    *status = CS_INVALID_HANDLE;
+    return;
+  }
+  Notifier::GetInstance().RemoveListener(uid);
 }
 
+bool NotifierDestroyed() { return Notifier::destroyed(); }
+
 //
 // Utility Functions
 //