[cscore] Add polled support to listener

Change Java VideoListener to use polling.
2026-06-19 00:41:43 +00:00 · 2021-01-26 23:07:16 -08:00
parent 483beb6361
commit 29bf9d6ef1
11 changed files with 445 additions and 260 deletions
--- a/cscore/src/main/java/edu/wpi/cscore/CameraServerJNI.java
+++ b/cscore/src/main/java/edu/wpi/cscore/CameraServerJNI.java
@@ -313,6 +313,19 @@ public class CameraServerJNI {

  public static native void removeListener(int handle);

+  public static native int createListenerPoller();
+
+  public static native void destroyListenerPoller(int poller);
+
+  public static native int addPolledListener(int poller, int eventMask, boolean immediateNotify);
+
+  public static native VideoEvent[] pollListener(int poller) throws InterruptedException;
+
+  public static native VideoEvent[] pollListenerTimeout(int poller, double timeout)
+      throws InterruptedException;
+
+  public static native void cancelPollListener(int poller);
+
  //
  // Telemetry Functions
  //
--- a/cscore/src/main/java/edu/wpi/cscore/VideoEvent.java
+++ b/cscore/src/main/java/edu/wpi/cscore/VideoEvent.java
@@ -102,7 +102,8 @@ public class VideoEvent {
      int property,
      int propertyKind,
      int value,
-      String valueStr) {
+      String valueStr,
+      int listener) {
    this.kind = getKindFromInt(kind);
    this.sourceHandle = source;
    this.sinkHandle = sink;
@@ -112,6 +113,7 @@ public class VideoEvent {
    this.propertyKind = VideoProperty.getKindFromInt(propertyKind);
    this.value = value;
    this.valueStr = valueStr;
+    this.listener = listener;
  }

  @SuppressWarnings("MemberName")
@@ -145,6 +147,10 @@ public class VideoEvent {
  @SuppressWarnings("MemberName")
  public String valueStr;

+  // Listener that was triggered
+  @SuppressWarnings("MemberName")
+  public int listener;
+
  public VideoSource getSource() {
    return new VideoSource(CameraServerJNI.copySource(sourceHandle));
  }
--- a/cscore/src/main/java/edu/wpi/cscore/VideoListener.java
+++ b/cscore/src/main/java/edu/wpi/cscore/VideoListener.java
@@ -4,6 +4,10 @@

 package edu.wpi.cscore;

+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;

 /**
@@ -20,15 +24,31 @@ public class VideoListener implements AutoCloseable {
   *     of events for the current library state.
   */
  public VideoListener(Consumer<VideoEvent> listener, int eventMask, boolean immediateNotify) {
-    m_handle = CameraServerJNI.addListener(listener, eventMask, immediateNotify);
+    s_lock.lock();
+    try {
+      if (s_poller == 0) {
+        s_poller = CameraServerJNI.createListenerPoller();
+        startThread();
+      }
+      m_handle = CameraServerJNI.addPolledListener(s_poller, eventMask, immediateNotify);
+      s_listeners.put(m_handle, listener);
+    } finally {
+      s_lock.unlock();
+    }
  }

  @Override
  public synchronized void close() {
    if (m_handle != 0) {
+      s_lock.lock();
+      try {
+        s_listeners.remove(m_handle);
+      } finally {
+        s_lock.unlock();
+      }
      CameraServerJNI.removeListener(m_handle);
+      m_handle = 0;
    }
-    m_handle = 0;
  }

  public boolean isValid() {
@@ -36,4 +56,71 @@ public class VideoListener implements AutoCloseable {
  }

  private int m_handle;
+
+  private static final ReentrantLock s_lock = new ReentrantLock();
+  private static final Map<Integer, Consumer<VideoEvent>> s_listeners = new HashMap<>();
+  private static Thread s_thread;
+  private static int s_poller;
+  private static boolean s_waitQueue;
+  private static final Condition s_waitQueueCond = s_lock.newCondition();
+
+  @SuppressWarnings("PMD.AvoidCatchingThrowable")
+  private static void startThread() {
+    s_thread =
+        new Thread(
+            () -> {
+              boolean wasInterrupted = false;
+              while (!Thread.interrupted()) {
+                VideoEvent[] events;
+                try {
+                  events = CameraServerJNI.pollListener(s_poller);
+                } catch (InterruptedException ex) {
+                  s_lock.lock();
+                  try {
+                    if (s_waitQueue) {
+                      s_waitQueue = false;
+                      s_waitQueueCond.signalAll();
+                      continue;
+                    }
+                  } finally {
+                    s_lock.unlock();
+                  }
+                  Thread.currentThread().interrupt();
+                  // don't try to destroy poller, as its handle is likely no longer valid
+                  wasInterrupted = true;
+                  break;
+                }
+                for (VideoEvent event : events) {
+                  Consumer<VideoEvent> listener;
+                  s_lock.lock();
+                  try {
+                    listener = s_listeners.get(event.listener);
+                  } finally {
+                    s_lock.unlock();
+                  }
+                  if (listener != null) {
+                    try {
+                      listener.accept(event);
+                    } catch (Throwable throwable) {
+                      System.err.println(
+                          "Unhandled exception during listener callback: " + throwable.toString());
+                      throwable.printStackTrace();
+                    }
+                  }
+                }
+              }
+              s_lock.lock();
+              try {
+                if (!wasInterrupted) {
+                  CameraServerJNI.destroyListenerPoller(s_poller);
+                }
+                s_poller = 0;
+              } finally {
+                s_lock.unlock();
+              }
+            },
+            "VideoListener");
+    s_thread.setDaemon(true);
+    s_thread.start();
+  }
 }
--- a/cscore/src/main/native/cpp/Handle.h
+++ b/cscore/src/main/native/cpp/Handle.h
@@ -22,7 +22,8 @@ class Handle {
    kSource,
    kSink,
    kListener,
-    kSinkProperty
+    kSinkProperty,
+    kListenerPoller
  };
  enum { kIndexMax = 0xffff };

--- a/cscore/src/main/native/cpp/Notifier.cpp
+++ b/cscore/src/main/native/cpp/Notifier.cpp
@@ -15,170 +15,26 @@

 using namespace cs;

-bool Notifier::s_destroyed = false;
+Notifier::Notifier() {}

-namespace {
-// Vector which provides an integrated freelist for removal and reuse of
-// individual elements.
-template <typename T>
-class UidVector {
- public:
-  using size_type = typename std::vector<T>::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;
-};
-
-}  // namespace
-
-class Notifier::Thread : public wpi::SafeThread {
- public:
-  Thread(std::function<void()> on_start, std::function<void()> on_exit)
-      : m_on_start(std::move(on_start)), m_on_exit(std::move(on_exit)) {}
-
-  void Main() override;
-
-  struct Listener {
-    Listener() = default;
-    Listener(std::function<void(const RawEvent& event)> callback_,
-             int eventMask_)
-        : callback(std::move(callback_)), eventMask(eventMask_) {}
-
-    explicit operator bool() const { return static_cast<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;
-}
+Notifier::~Notifier() {}

 void Notifier::Start() {
-  m_owner.Start(m_on_start, m_on_exit);
+  DoStart();
 }

-void Notifier::Stop() {
-  m_owner.Stop();
+unsigned int Notifier::Add(std::function<void(const RawEvent& event)> callback,
+                           int eventMask) {
+  return DoAdd(callback, eventMask);
 }

-void Notifier::Thread::Main() {
-  if (m_on_start) {
-    m_on_start();
-  }
-
-  std::unique_lock 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 (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.kind & 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);
+unsigned int Notifier::AddPolled(unsigned int pollerUid, int eventMask) {
+  return DoAdd(pollerUid, eventMask);
 }

 void Notifier::NotifySource(const wpi::Twine& name, CS_Source source,
                            CS_EventKind kind) {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-  thr->m_notifications.emplace(name, source, static_cast<RawEvent::Kind>(kind));
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, name, source, static_cast<RawEvent::Kind>(kind));
 }

 void Notifier::NotifySource(const SourceImpl& source, CS_EventKind kind) {
@@ -188,44 +44,24 @@ void Notifier::NotifySource(const SourceImpl& source, CS_EventKind kind) {

 void Notifier::NotifySourceVideoMode(const SourceImpl& source,
                                     const VideoMode& mode) {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
  auto handleData = Instance::GetInstance().FindSource(source);
-
-  thr->m_notifications.emplace(source.GetName(), handleData.first, mode);
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, source.GetName(), handleData.first, mode);
 }

 void Notifier::NotifySourceProperty(const SourceImpl& source, CS_EventKind kind,
                                    const wpi::Twine& propertyName,
                                    int property, CS_PropertyKind propertyKind,
                                    int value, const wpi::Twine& valueStr) {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
  auto handleData = Instance::GetInstance().FindSource(source);
-
-  thr->m_notifications.emplace(
-      propertyName, handleData.first, static_cast<RawEvent::Kind>(kind),
-      Handle{handleData.first, property, Handle::kProperty}, propertyKind,
-      value, valueStr);
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, propertyName, handleData.first,
+       static_cast<RawEvent::Kind>(kind),
+       Handle{handleData.first, property, Handle::kProperty}, propertyKind,
+       value, valueStr);
 }

 void Notifier::NotifySink(const wpi::Twine& name, CS_Sink sink,
                          CS_EventKind kind) {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
-  thr->m_notifications.emplace(name, sink, static_cast<RawEvent::Kind>(kind));
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, name, sink, static_cast<RawEvent::Kind>(kind));
 }

 void Notifier::NotifySink(const SinkImpl& sink, CS_EventKind kind) {
@@ -235,52 +71,26 @@ void Notifier::NotifySink(const SinkImpl& sink, CS_EventKind kind) {

 void Notifier::NotifySinkSourceChanged(const wpi::Twine& name, CS_Sink sink,
                                       CS_Source source) {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
  RawEvent event{name, sink, RawEvent::kSinkSourceChanged};
  event.sourceHandle = source;
-
-  thr->m_notifications.emplace(std::move(event));
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, std::move(event));
 }

 void Notifier::NotifySinkProperty(const SinkImpl& sink, CS_EventKind kind,
                                  const wpi::Twine& propertyName, int property,
                                  CS_PropertyKind propertyKind, int value,
                                  const wpi::Twine& valueStr) {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
  auto handleData = Instance::GetInstance().FindSink(sink);
-
-  thr->m_notifications.emplace(
-      propertyName, handleData.first, static_cast<RawEvent::Kind>(kind),
-      Handle{handleData.first, property, Handle::kSinkProperty}, propertyKind,
-      value, valueStr);
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, propertyName, handleData.first,
+       static_cast<RawEvent::Kind>(kind),
+       Handle{handleData.first, property, Handle::kSinkProperty}, propertyKind,
+       value, valueStr);
 }

 void Notifier::NotifyNetworkInterfacesChanged() {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
-  thr->m_notifications.emplace(RawEvent::kNetworkInterfacesChanged);
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, RawEvent::kNetworkInterfacesChanged);
 }

 void Notifier::NotifyTelemetryUpdated() {
-  auto thr = m_owner.GetThread();
-  if (!thr) {
-    return;
-  }
-
-  thr->m_notifications.emplace(RawEvent::kTelemetryUpdated);
-  thr->m_cond.notify_one();
+  Send(UINT_MAX, RawEvent::kTelemetryUpdated);
 }
--- a/cscore/src/main/native/cpp/Notifier.h
+++ b/cscore/src/main/native/cpp/Notifier.h
@@ -6,9 +6,11 @@
 #define CSCORE_NOTIFIER_H_

 #include <functional>
+#include <utility>

-#include <wpi/SafeThread.h>
+#include <wpi/CallbackManager.h>

+#include "Handle.h"
 #include "cscore_cpp.h"

 namespace cs {
@@ -16,7 +18,40 @@ namespace cs {
 class SinkImpl;
 class SourceImpl;

-class Notifier {
+namespace impl {
+
+struct ListenerData : public wpi::CallbackListenerData<
+                          std::function<void(const RawEvent& event)>> {
+  ListenerData() = default;
+  ListenerData(std::function<void(const RawEvent& event)> callback_,
+               int eventMask_)
+      : CallbackListenerData(std::move(callback_)), eventMask(eventMask_) {}
+  ListenerData(unsigned int pollerUid_, int eventMask_)
+      : CallbackListenerData(pollerUid_), eventMask(eventMask_) {}
+
+  int eventMask;
+};
+
+class NotifierThread
+    : public wpi::CallbackThread<NotifierThread, RawEvent, ListenerData> {
+ public:
+  bool Matches(const ListenerData& /*listener*/, const RawEvent& /*data*/) {
+    return true;
+  }
+
+  void SetListener(RawEvent* data, unsigned int listener_uid) {
+    data->listener = Handle(listener_uid, Handle::kListener);
+  }
+
+  void DoCallback(std::function<void(const RawEvent& event)> callback,
+                  const RawEvent& data) {
+    callback(data);
+  }
+};
+
+}  // namespace impl
+
+class Notifier : public wpi::CallbackManager<Notifier, impl::NotifierThread> {
  friend class NotifierTest;

 public:
@@ -24,16 +59,10 @@ class Notifier {
  ~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);
+  unsigned int Add(std::function<void(const RawEvent& event)> callback,
+                   int eventMask);
+  unsigned int AddPolled(unsigned int pollerUid, int eventMask);

  // Notification events
  void NotifySource(const wpi::Twine& name, CS_Source source,
@@ -54,14 +83,6 @@ class Notifier {
                          const wpi::Twine& valueStr);
  void NotifyNetworkInterfacesChanged();
  void NotifyTelemetryUpdated();
-
- private:
-  class Thread;
-  wpi::SafeThreadOwner<Thread> m_owner;
-
-  std::function<void()> m_on_start;
-  std::function<void()> m_on_exit;
-  static bool s_destroyed;
 };

 }  // namespace cs
--- a/cscore/src/main/native/cpp/cscore_c.cpp
+++ b/cscore/src/main/native/cpp/cscore_c.cpp
@@ -15,6 +15,39 @@
 #include "cscore_cpp.h"
 #include "cscore_raw.h"

+static CS_Event ConvertToC(const cs::RawEvent& rawEvent) {
+  CS_Event event;
+  event.kind = static_cast<CS_EventKind>(static_cast<int>(rawEvent.kind));
+  event.source = rawEvent.sourceHandle;
+  event.sink = rawEvent.sinkHandle;
+  event.name = rawEvent.name.c_str();
+  event.mode = rawEvent.mode;
+  event.property = rawEvent.propertyHandle;
+  event.propertyKind = rawEvent.propertyKind;
+  event.value = rawEvent.value;
+  event.valueStr = rawEvent.valueStr.c_str();
+  event.listener = rawEvent.listener;
+  return event;
+}
+
+template <typename O, typename I>
+static O* ConvertToC(std::vector<I>&& in, int* count) {
+  using T = std::vector<I>;
+  size_t size = in.size();
+  O* out = static_cast<O*>(wpi::safe_malloc(size * sizeof(O) + sizeof(T)));
+  *count = size;
+  for (size_t i = 0; i < size; ++i) {
+    out[i] = ConvertToC(in[i]);
+  }
+
+  // retain vector at end of returned array
+  alignas(T) unsigned char buf[sizeof(T)];
+  new (buf) T(std::move(in));
+  std::memcpy(out + size * sizeof(O), buf, sizeof(T));
+
+  return out;
+}
+
 extern "C" {

 CS_PropertyKind CS_GetPropertyKind(CS_Property property, CS_Status* status) {
@@ -332,16 +365,7 @@ CS_Listener CS_AddListener(void* data,
                           CS_Status* status) {
  return cs::AddListener(
      [=](const cs::RawEvent& rawEvent) {
-        CS_Event event;
-        event.kind = static_cast<CS_EventKind>(static_cast<int>(rawEvent.kind));
-        event.source = rawEvent.sourceHandle;
-        event.sink = rawEvent.sinkHandle;
-        event.name = rawEvent.name.c_str();
-        event.mode = rawEvent.mode;
-        event.property = rawEvent.propertyHandle;
-        event.propertyKind = rawEvent.propertyKind;
-        event.value = rawEvent.value;
-        event.valueStr = rawEvent.valueStr.c_str();
+        CS_Event event = ConvertToC(rawEvent);
        callback(data, &event);
      },
      eventMask, immediateNotify, status);
@@ -351,6 +375,45 @@ void CS_RemoveListener(CS_Listener handle, CS_Status* status) {
  return cs::RemoveListener(handle, status);
 }

+CS_ListenerPoller CS_CreateListenerPoller(void) {
+  return cs::CreateListenerPoller();
+}
+
+void CS_DestroyListenerPoller(CS_ListenerPoller poller) {
+  cs::DestroyListenerPoller(poller);
+}
+
+CS_Listener CS_AddPolledListener(CS_ListenerPoller poller, int eventMask,
+                                 CS_Bool immediateNotify, CS_Status* status) {
+  return cs::AddPolledListener(poller, eventMask, immediateNotify, status);
+}
+
+struct CS_Event* CS_PollListener(CS_ListenerPoller poller, int* count) {
+  return ConvertToC<CS_Event>(cs::PollListener(poller), count);
+}
+
+struct CS_Event* CS_PollListenerTimeout(CS_ListenerPoller poller, int* count,
+                                        double timeout, CS_Bool* timedOut) {
+  bool cppTimedOut = false;
+  auto arrCpp = cs::PollListener(poller, timeout, &cppTimedOut);
+  *timedOut = cppTimedOut;
+  return ConvertToC<CS_Event>(std::move(arrCpp), count);
+}
+
+void CS_CancelPollListener(CS_ListenerPoller poller) {
+  cs::CancelPollListener(poller);
+}
+
+void CS_FreeEvents(CS_Event* arr, int count) {
+  // destroy vector saved at end of array
+  using T = std::vector<cs::RawEvent>;
+  alignas(T) unsigned char buf[sizeof(T)];
+  std::memcpy(buf, arr + count * sizeof(CS_Event), sizeof(T));
+  reinterpret_cast<T*>(buf)->~T();
+
+  std::free(arr);
+}
+
 int CS_NotifierDestroyed(void) {
  return cs::NotifierDestroyed();
 }
--- a/cscore/src/main/native/cpp/cscore_cpp.cpp
+++ b/cscore/src/main/native/cpp/cscore_cpp.cpp
@@ -705,19 +705,12 @@ void ReleaseSink(CS_Sink sink, CS_Status* status) {
 // Listener Functions
 //

-void SetListenerOnStart(std::function<void()> onStart) {
-  Instance::GetInstance().notifier.SetOnStart(onStart);
-}
+void SetListenerOnStart(std::function<void()> onStart) {}

-void SetListenerOnExit(std::function<void()> onExit) {
-  Instance::GetInstance().notifier.SetOnExit(onExit);
-}
+void SetListenerOnExit(std::function<void()> onExit) {}

-CS_Listener AddListener(std::function<void(const RawEvent& event)> callback,
-                        int eventMask, bool immediateNotify,
-                        CS_Status* status) {
+static void StartBackground(int eventMask, bool immediateNotify) {
  auto& inst = Instance::GetInstance();
-  int uid = inst.notifier.AddListener(callback, eventMask);
  if ((eventMask & CS_NETWORK_INTERFACES_CHANGED) != 0) {
    // start network interface event listener
    inst.networkListener.Start();
@@ -725,6 +718,14 @@ CS_Listener AddListener(std::function<void(const RawEvent& event)> callback,
      inst.notifier.NotifyNetworkInterfacesChanged();
    }
  }
+}
+
+CS_Listener AddListener(std::function<void(const RawEvent& event)> callback,
+                        int eventMask, bool immediateNotify,
+                        CS_Status* status) {
+  auto& inst = Instance::GetInstance();
+  int uid = inst.notifier.Add(callback, eventMask);
+  StartBackground(eventMask, immediateNotify);
  if (immediateNotify) {
    // TODO
  }
@@ -737,11 +738,67 @@ void RemoveListener(CS_Listener handle, CS_Status* status) {
    *status = CS_INVALID_HANDLE;
    return;
  }
-  Instance::GetInstance().notifier.RemoveListener(uid);
+  Instance::GetInstance().notifier.Remove(uid);
+}
+
+CS_ListenerPoller CreateListenerPoller() {
+  auto& inst = Instance::GetInstance();
+  return Handle(inst.notifier.CreatePoller(), Handle::kListenerPoller);
+}
+
+void DestroyListenerPoller(CS_ListenerPoller poller) {
+  int uid = Handle{poller}.GetTypedIndex(Handle::kListenerPoller);
+  if (uid < 0) {
+    return;
+  }
+  Instance::GetInstance().notifier.RemovePoller(uid);
+}
+
+CS_Listener AddPolledListener(CS_ListenerPoller poller, int eventMask,
+                              bool immediateNotify, CS_Status* status) {
+  Handle handle{poller};
+  int id = handle.GetTypedIndex(Handle::kListenerPoller);
+  if (id < 0) {
+    *status = CS_INVALID_HANDLE;
+    return 0;
+  }
+
+  auto& inst = Instance::GetInstance();
+  int uid = inst.notifier.AddPolled(id, eventMask);
+  StartBackground(eventMask, immediateNotify);
+  return Handle{uid, Handle::kListener};
+}
+
+std::vector<RawEvent> PollListener(CS_ListenerPoller poller) {
+  Handle handle{poller};
+  int id = handle.GetTypedIndex(Handle::kListenerPoller);
+  if (id < 0) {
+    return {};
+  }
+  return Instance::GetInstance().notifier.Poll(id);
+}
+
+std::vector<RawEvent> PollListener(CS_ListenerPoller poller, double timeout,
+                                   bool* timedOut) {
+  Handle handle{poller};
+  int id = handle.GetTypedIndex(Handle::kListenerPoller);
+  if (id < 0) {
+    return {};
+  }
+  return Instance::GetInstance().notifier.Poll(id, timeout, timedOut);
+}
+
+void CancelPollListener(CS_ListenerPoller poller) {
+  Handle handle{poller};
+  int id = handle.GetTypedIndex(Handle::kListenerPoller);
+  if (id < 0) {
+    return;
+  }
+  return Instance::GetInstance().notifier.CancelPoll(id);
 }

 bool NotifierDestroyed() {
-  return Notifier::destroyed();
+  return false;
 }

 //
--- a/cscore/src/main/native/cpp/jni/CameraServerJNI.cpp
+++ b/cscore/src/main/native/cpp/jni/CameraServerJNI.cpp
@@ -31,6 +31,7 @@ static JClass videoModeCls;
 static JClass videoEventCls;
 static JClass rawFrameCls;
 static JException videoEx;
+static JException interruptedEx;
 static JException nullPointerEx;
 static JException unsupportedEx;
 static JException exceptionEx;
@@ -45,6 +46,7 @@ static const JClassInit classes[] = {

 static const JExceptionInit exceptions[] = {
    {"edu/wpi/cscore/VideoException", &videoEx},
+    {"java/lang/InterruptedException", &interruptedEx},
    {"java/lang/NullPointerException", &nullPointerEx},
    {"java/lang/UnsupportedOperationException", &unsupportedEx},
    {"java/lang/Exception", &exceptionEx}};
@@ -268,7 +270,7 @@ static jobject MakeJObject(JNIEnv* env, const cs::VideoMode& videoMode) {
 static jobject MakeJObject(JNIEnv* env, const cs::RawEvent& event) {
  static jmethodID constructor =
      env->GetMethodID(videoEventCls, "<init>",
-                       "(IIILjava/lang/String;IIIIIIILjava/lang/String;)V");
+                       "(IIILjava/lang/String;IIIIIIILjava/lang/String;I)V");
  JLocal<jstring> name(env, MakeJString(env, event.name));
  JLocal<jstring> valueStr(env, MakeJString(env, event.valueStr));
  // clang-format off
@@ -286,10 +288,23 @@ static jobject MakeJObject(JNIEnv* env, const cs::RawEvent& event) {
      static_cast<jint>(event.propertyHandle),
      static_cast<jint>(event.propertyKind),
      static_cast<jint>(event.value),
-      valueStr.obj());
+      valueStr.obj(),
+      static_cast<jint>(event.listener));
  // clang-format on
 }

+static jobjectArray MakeJObject(JNIEnv* env, wpi::ArrayRef<cs::RawEvent> arr) {
+  jobjectArray jarr = env->NewObjectArray(arr.size(), videoEventCls, nullptr);
+  if (!jarr) {
+    return nullptr;
+  }
+  for (size_t i = 0; i < arr.size(); ++i) {
+    JLocal<jobject> elem{env, MakeJObject(env, arr[i])};
+    env->SetObjectArrayElement(jarr, i, elem.obj());
+  }
+  return jarr;
+}
+
 extern "C" {

 /*
@@ -1881,6 +1896,92 @@ Java_edu_wpi_cscore_CameraServerJNI_removeListener
  CheckStatus(env, status);
 }

+/*
+ * Class:     edu_wpi_cscore_CameraServerJNI
+ * Method:    createListenerPoller
+ * Signature: ()I
+ */
+JNIEXPORT jint JNICALL
+Java_edu_wpi_cscore_CameraServerJNI_createListenerPoller
+  (JNIEnv*, jclass)
+{
+  return cs::CreateListenerPoller();
+}
+
+/*
+ * Class:     edu_wpi_cscore_CameraServerJNI
+ * Method:    destroyListenerPoller
+ * Signature: (I)V
+ */
+JNIEXPORT void JNICALL
+Java_edu_wpi_cscore_CameraServerJNI_destroyListenerPoller
+  (JNIEnv*, jclass, jint poller)
+{
+  cs::DestroyListenerPoller(poller);
+}
+
+/*
+ * Class:     edu_wpi_cscore_CameraServerJNI
+ * Method:    addPolledListener
+ * Signature: (IIZ)I
+ */
+JNIEXPORT jint JNICALL
+Java_edu_wpi_cscore_CameraServerJNI_addPolledListener
+  (JNIEnv* env, jclass, jint poller, jint eventMask, jboolean immediateNotify)
+{
+  CS_Status status = 0;
+  auto rv = cs::AddPolledListener(poller, eventMask, immediateNotify, &status);
+  CheckStatus(env, status);
+  return rv;
+}
+
+/*
+ * Class:     edu_wpi_cscore_CameraServerJNI
+ * Method:    pollListener
+ * Signature: (I)[Ljava/lang/Object;
+ */
+JNIEXPORT jobjectArray JNICALL
+Java_edu_wpi_cscore_CameraServerJNI_pollListener
+  (JNIEnv* env, jclass, jint poller)
+{
+  auto events = cs::PollListener(poller);
+  if (events.empty()) {
+    interruptedEx.Throw(env, "PollListener interrupted");
+    return nullptr;
+  }
+  return MakeJObject(env, events);
+}
+
+/*
+ * Class:     edu_wpi_cscore_CameraServerJNI
+ * Method:    pollListenerTimeout
+ * Signature: (ID)[Ljava/lang/Object;
+ */
+JNIEXPORT jobjectArray JNICALL
+Java_edu_wpi_cscore_CameraServerJNI_pollListenerTimeout
+  (JNIEnv* env, jclass, jint poller, jdouble timeout)
+{
+  bool timed_out = false;
+  auto events = cs::PollListener(poller, timeout, &timed_out);
+  if (events.empty() && !timed_out) {
+    interruptedEx.Throw(env, "PollListener interrupted");
+    return nullptr;
+  }
+  return MakeJObject(env, events);
+}
+
+/*
+ * Class:     edu_wpi_cscore_CameraServerJNI
+ * Method:    cancelPollListener
+ * Signature: (I)V
+ */
+JNIEXPORT void JNICALL
+Java_edu_wpi_cscore_CameraServerJNI_cancelPollListener
+  (JNIEnv*, jclass, jint poller)
+{
+  cs::CancelPollListener(poller);
+}
+
 /*
 * Class:     edu_wpi_cscore_CameraServerJNI
 * Method:    setTelemetryPeriod
--- a/cscore/src/main/native/include/cscore_c.h
+++ b/cscore/src/main/native/include/cscore_c.h
@@ -46,6 +46,7 @@ typedef int CS_Status;
 typedef int CS_Handle;
 typedef CS_Handle CS_Property;
 typedef CS_Handle CS_Listener;
+typedef CS_Handle CS_ListenerPoller;
 typedef CS_Handle CS_Sink;
 typedef CS_Handle CS_Source;
 /** @} */
@@ -222,6 +223,9 @@ struct CS_Event {
  enum CS_PropertyKind propertyKind;
  int value;
  const char* valueStr;
+
+  // Listener that was triggered
+  CS_Listener listener;
 };

 /**
@@ -429,9 +433,19 @@ void CS_SetListenerOnStart(void (*onStart)(void* data), void* data);
 void CS_SetListenerOnExit(void (*onExit)(void* data), void* data);
 CS_Listener CS_AddListener(
    void* data, void (*callback)(void* data, const struct CS_Event* event),
-    int eventMask, int immediateNotify, CS_Status* status);
+    int eventMask, CS_Bool immediateNotify, CS_Status* status);

 void CS_RemoveListener(CS_Listener handle, CS_Status* status);
+
+CS_ListenerPoller CS_CreateListenerPoller(void);
+void CS_DestroyListenerPoller(CS_ListenerPoller poller);
+CS_Listener CS_AddPolledListener(CS_ListenerPoller poller, int eventMask,
+                                 CS_Bool immediateNotify, CS_Status* status);
+struct CS_Event* CS_PollListener(CS_ListenerPoller poller, int* count);
+struct CS_Event* CS_PollListenerTimeout(CS_ListenerPoller poller, int* count,
+                                        double timeout, CS_Bool* timedOut);
+void CS_FreeEvents(struct CS_Event* arr, int count);
+void CS_CancelPollListener(CS_ListenerPoller poller);
 /** @} */

 int CS_NotifierDestroyed(void);
--- a/cscore/src/main/native/include/cscore_cpp.h
+++ b/cscore/src/main/native/include/cscore_cpp.h
@@ -163,6 +163,9 @@ struct RawEvent {
  CS_PropertyKind propertyKind;
  int value;
  std::string valueStr;
+
+  // Listener that was triggered
+  CS_Listener listener{0};
 };

 /**
@@ -378,6 +381,15 @@ CS_Listener AddListener(std::function<void(const RawEvent& event)> callback,
                        int eventMask, bool immediateNotify, CS_Status* status);

 void RemoveListener(CS_Listener handle, CS_Status* status);
+
+CS_ListenerPoller CreateListenerPoller();
+void DestroyListenerPoller(CS_ListenerPoller poller);
+CS_Listener AddPolledListener(CS_ListenerPoller poller, int eventMask,
+                              bool immediateNotify, CS_Status* status);
+std::vector<RawEvent> PollListener(CS_ListenerPoller poller);
+std::vector<RawEvent> PollListener(CS_ListenerPoller poller, double timeout,
+                                   bool* timedOut);
+void CancelPollListener(CS_ListenerPoller poller);
 /** @} */

 bool NotifierDestroyed();