[cscore] Fix Java direct callback notifications (#3631)

These relied on the OnStart and OnExit callbacks which were not present
in the wpiutil common CallbackManager code.  Add support for these and
fix up other use cases.

Also fixes notifications to correctly filter on event kind.

This fixes a breakage caused by #3133.

cscore/src/main/native/cpp/Notifier.h

@@ -35,8 +35,11 @@ struct ListenerData : public wpi::CallbackListenerData<
 class NotifierThread
     : public wpi::CallbackThread<NotifierThread, RawEvent, ListenerData> {
  public:
-  bool Matches(const ListenerData& /*listener*/, const RawEvent& /*data*/) {
-    return true;
+  NotifierThread(std::function<void()> on_start, std::function<void()> on_exit)
+      : CallbackThread(std::move(on_start), std::move(on_exit)) {}
+
+  bool Matches(const ListenerData& listener, const RawEvent& data) {
+    return (data.kind & listener.eventMask) != 0;
   }
 
   void SetListener(RawEvent* data, unsigned int listener_uid) {

cscore/src/main/native/cpp/cscore_cpp.cpp

@@ -708,9 +708,13 @@ void ReleaseSink(CS_Sink sink, CS_Status* status) {
 // Listener Functions
 //
 
-void SetListenerOnStart(std::function<void()> onStart) {}
+void SetListenerOnStart(std::function<void()> onStart) {
+  Instance::GetInstance().notifier.SetOnStart(onStart);
+}
 
-void SetListenerOnExit(std::function<void()> onExit) {}
+void SetListenerOnExit(std::function<void()> onExit) {
+  Instance::GetInstance().notifier.SetOnExit(onExit);
+}
 
 static void StartBackground(int eventMask, bool immediateNotify) {
   auto& inst = Instance::GetInstance();

ntcore/src/main/native/cpp/ConnectionNotifier.h

@@ -5,6 +5,8 @@
 #ifndef NTCORE_CONNECTIONNOTIFIER_H_
 #define NTCORE_CONNECTIONNOTIFIER_H_
 
+#include <utility>
+
 #include <wpi/CallbackManager.h>
 
 #include "Handle.h"
@@ -19,7 +21,9 @@ class ConnectionNotifierThread
     : public wpi::CallbackThread<ConnectionNotifierThread,
                                  ConnectionNotification> {
  public:
-  explicit ConnectionNotifierThread(int inst) : m_inst(inst) {}
+  ConnectionNotifierThread(std::function<void()> on_start,
+                           std::function<void()> on_exit, int inst)
+      : CallbackThread(std::move(on_start), std::move(on_exit)), m_inst(inst) {}
 
   bool Matches(const ListenerData& /*listener*/,
                const ConnectionNotification& /*data*/) {

ntcore/src/main/native/cpp/EntryNotifier.h

@@ -9,6 +9,7 @@
 #include <memory>
 #include <string>
 #include <string_view>
+#include <utility>
 
 #include <wpi/CallbackManager.h>
 
@@ -52,7 +53,9 @@ class EntryNotifierThread
     : public wpi::CallbackThread<EntryNotifierThread, EntryNotification,
                                  EntryListenerData> {
  public:
-  explicit EntryNotifierThread(int inst) : m_inst(inst) {}
+  EntryNotifierThread(std::function<void()> on_start,
+                      std::function<void()> on_exit, int inst)
+      : CallbackThread(std::move(on_start), std::move(on_exit)), m_inst(inst) {}
 
   bool Matches(const EntryListenerData& listener,
                const EntryNotification& data);

ntcore/src/main/native/cpp/LoggerImpl.h

@@ -5,6 +5,8 @@
 #ifndef NTCORE_LOGGERIMPL_H_
 #define NTCORE_LOGGERIMPL_H_
 
+#include <utility>
+
 #include <wpi/CallbackManager.h>
 
 #include "Handle.h"
@@ -35,7 +37,9 @@ struct LoggerListenerData : public wpi::CallbackListenerData<
 class LoggerThread
     : public wpi::CallbackThread<LoggerThread, LogMessage, LoggerListenerData> {
  public:
-  explicit LoggerThread(int inst) : m_inst(inst) {}
+  LoggerThread(std::function<void()> on_start, std::function<void()> on_exit,
+               int inst)
+      : CallbackThread(std::move(on_start), std::move(on_exit)), m_inst(inst) {}
 
   bool Matches(const LoggerListenerData& listener, const LogMessage& data) {
     return data.level >= listener.min_level && data.level <= listener.max_level;

ntcore/src/main/native/cpp/RpcServer.h

@@ -38,8 +38,11 @@ class RpcServerThread
     : public wpi::CallbackThread<RpcServerThread, RpcAnswer, RpcListenerData,
                                  RpcNotifierData> {
  public:
-  RpcServerThread(int inst, wpi::Logger& logger)
-      : m_inst(inst), m_logger(logger) {}
+  RpcServerThread(std::function<void()> on_start, std::function<void()> on_exit,
+                  int inst, wpi::Logger& logger)
+      : CallbackThread(std::move(on_start), std::move(on_exit)),
+        m_inst(inst),
+        m_logger(logger) {}
 
   bool Matches(const RpcListenerData& /*listener*/,
                const RpcNotifierData& data) {

wpiutil/src/main/native/include/wpi/CallbackManager.h

@@ -53,6 +53,9 @@ class CallbackThread : public wpi::SafeThread {
   using NotifierData = TNotifierData;
   using ListenerData = TListenerData;
 
+  CallbackThread(std::function<void()> on_start, std::function<void()> on_exit)
+      : m_on_start(std::move(on_start)), m_on_exit(std::move(on_exit)) {}
+
   ~CallbackThread() override {
     // Wake up any blocked pollers
     for (size_t i = 0; i < m_pollers.size(); ++i) {
@@ -85,6 +88,9 @@ class CallbackThread : public wpi::SafeThread {
   };
   wpi::UidVector<std::shared_ptr<Poller>, 64> m_pollers;
 
+  std::function<void()> m_on_start;
+  std::function<void()> m_on_exit;
+
   // Must be called with m_mutex held
   template <typename... Args>
   void SendPoller(unsigned int poller_uid, Args&&... args) {
@@ -106,18 +112,22 @@ class CallbackThread : public wpi::SafeThread {
 template <typename Derived, typename TUserInfo, typename TListenerData,
           typename TNotifierData>
 void CallbackThread<Derived, TUserInfo, TListenerData, TNotifierData>::Main() {
+  if (m_on_start) {
+    m_on_start();
+  }
+
   std::unique_lock lock(m_mutex);
   while (m_active) {
     while (m_queue.empty()) {
       m_cond.wait(lock);
       if (!m_active) {
-        return;
+        goto done;
       }
     }
 
     while (!m_queue.empty()) {
       if (!m_active) {
-        return;
+        goto done;
       }
       auto item = std::move(m_queue.front());
 
@@ -144,6 +154,7 @@ void CallbackThread<Derived, TUserInfo, TListenerData, TNotifierData>::Main() {
           if (!listener) {
             continue;
           }
+
           if (!static_cast<Derived*>(this)->Matches(listener, item.second)) {
             continue;
           }
@@ -164,6 +175,11 @@ void CallbackThread<Derived, TUserInfo, TListenerData, TNotifierData>::Main() {
 
     m_queue_empty.notify_all();
   }
+
+done:
+  if (m_on_exit) {
+    m_on_exit();
+  }
 }
 
 // CRTP callback manager
@@ -177,6 +193,14 @@ class CallbackManager {
   friend class RpcServerTest;
 
  public:
+  void SetOnStart(std::function<void()> on_start) {
+    m_on_start = std::move(on_start);
+  }
+
+  void SetOnExit(std::function<void()> on_exit) {
+    m_on_exit = std::move(on_exit);
+  }
+
   void Stop() { m_owner.Stop(); }
 
   void Remove(unsigned int listener_uid) {
@@ -333,7 +357,7 @@ class CallbackManager {
  protected:
   template <typename... Args>
   void DoStart(Args&&... args) {
-    m_owner.Start(std::forward<Args>(args)...);
+    m_owner.Start(m_on_start, m_on_exit, std::forward<Args>(args)...);
   }
 
   template <typename... Args>
@@ -361,6 +385,9 @@ class CallbackManager {
 
  private:
   wpi::SafeThreadOwner<Thread> m_owner;
+
+  std::function<void()> m_on_start;
+  std::function<void()> m_on_exit;
 };
 
 }  // namespace wpi