From 61996c2bbf580c9e33fcfe1d6cc0456c03aa0265 Mon Sep 17 00:00:00 2001 From: Peter Johnson Date: Wed, 13 Oct 2021 08:46:07 -0700 Subject: [PATCH] [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 | 7 ++-- cscore/src/main/native/cpp/cscore_cpp.cpp | 8 +++-- .../src/main/native/cpp/ConnectionNotifier.h | 6 +++- ntcore/src/main/native/cpp/EntryNotifier.h | 5 ++- ntcore/src/main/native/cpp/LoggerImpl.h | 6 +++- ntcore/src/main/native/cpp/RpcServer.h | 7 ++-- .../main/native/include/wpi/CallbackManager.h | 33 +++++++++++++++++-- 7 files changed, 60 insertions(+), 12 deletions(-) diff --git a/cscore/src/main/native/cpp/Notifier.h b/cscore/src/main/native/cpp/Notifier.h index c0f7abcf39..c28e25c5a1 100644 --- a/cscore/src/main/native/cpp/Notifier.h +++ b/cscore/src/main/native/cpp/Notifier.h @@ -35,8 +35,11 @@ struct ListenerData : public wpi::CallbackListenerData< class NotifierThread : public wpi::CallbackThread { public: - bool Matches(const ListenerData& /*listener*/, const RawEvent& /*data*/) { - return true; + NotifierThread(std::function on_start, std::function 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) { diff --git a/cscore/src/main/native/cpp/cscore_cpp.cpp b/cscore/src/main/native/cpp/cscore_cpp.cpp index 4faae795ed..3b4e5703bc 100644 --- a/cscore/src/main/native/cpp/cscore_cpp.cpp +++ b/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 onStart) {} +void SetListenerOnStart(std::function onStart) { + Instance::GetInstance().notifier.SetOnStart(onStart); +} -void SetListenerOnExit(std::function onExit) {} +void SetListenerOnExit(std::function onExit) { + Instance::GetInstance().notifier.SetOnExit(onExit); +} static void StartBackground(int eventMask, bool immediateNotify) { auto& inst = Instance::GetInstance(); diff --git a/ntcore/src/main/native/cpp/ConnectionNotifier.h b/ntcore/src/main/native/cpp/ConnectionNotifier.h index 66dd5f1bc2..fab47332f1 100644 --- a/ntcore/src/main/native/cpp/ConnectionNotifier.h +++ b/ntcore/src/main/native/cpp/ConnectionNotifier.h @@ -5,6 +5,8 @@ #ifndef NTCORE_CONNECTIONNOTIFIER_H_ #define NTCORE_CONNECTIONNOTIFIER_H_ +#include + #include #include "Handle.h" @@ -19,7 +21,9 @@ class ConnectionNotifierThread : public wpi::CallbackThread { public: - explicit ConnectionNotifierThread(int inst) : m_inst(inst) {} + ConnectionNotifierThread(std::function on_start, + std::function on_exit, int inst) + : CallbackThread(std::move(on_start), std::move(on_exit)), m_inst(inst) {} bool Matches(const ListenerData& /*listener*/, const ConnectionNotification& /*data*/) { diff --git a/ntcore/src/main/native/cpp/EntryNotifier.h b/ntcore/src/main/native/cpp/EntryNotifier.h index 087d39321d..bbe2172b2e 100644 --- a/ntcore/src/main/native/cpp/EntryNotifier.h +++ b/ntcore/src/main/native/cpp/EntryNotifier.h @@ -9,6 +9,7 @@ #include #include #include +#include #include @@ -52,7 +53,9 @@ class EntryNotifierThread : public wpi::CallbackThread { public: - explicit EntryNotifierThread(int inst) : m_inst(inst) {} + EntryNotifierThread(std::function on_start, + std::function on_exit, int inst) + : CallbackThread(std::move(on_start), std::move(on_exit)), m_inst(inst) {} bool Matches(const EntryListenerData& listener, const EntryNotification& data); diff --git a/ntcore/src/main/native/cpp/LoggerImpl.h b/ntcore/src/main/native/cpp/LoggerImpl.h index 4f341838dc..2b577c1199 100644 --- a/ntcore/src/main/native/cpp/LoggerImpl.h +++ b/ntcore/src/main/native/cpp/LoggerImpl.h @@ -5,6 +5,8 @@ #ifndef NTCORE_LOGGERIMPL_H_ #define NTCORE_LOGGERIMPL_H_ +#include + #include #include "Handle.h" @@ -35,7 +37,9 @@ struct LoggerListenerData : public wpi::CallbackListenerData< class LoggerThread : public wpi::CallbackThread { public: - explicit LoggerThread(int inst) : m_inst(inst) {} + LoggerThread(std::function on_start, std::function 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; diff --git a/ntcore/src/main/native/cpp/RpcServer.h b/ntcore/src/main/native/cpp/RpcServer.h index dbf2154ea8..b8ae6b7bad 100644 --- a/ntcore/src/main/native/cpp/RpcServer.h +++ b/ntcore/src/main/native/cpp/RpcServer.h @@ -38,8 +38,11 @@ class RpcServerThread : public wpi::CallbackThread { public: - RpcServerThread(int inst, wpi::Logger& logger) - : m_inst(inst), m_logger(logger) {} + RpcServerThread(std::function on_start, std::function 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) { diff --git a/wpiutil/src/main/native/include/wpi/CallbackManager.h b/wpiutil/src/main/native/include/wpi/CallbackManager.h index d018fc1c3b..32e2adddc7 100644 --- a/wpiutil/src/main/native/include/wpi/CallbackManager.h +++ b/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 on_start, std::function 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, 64> m_pollers; + std::function m_on_start; + std::function m_on_exit; + // Must be called with m_mutex held template void SendPoller(unsigned int poller_uid, Args&&... args) { @@ -106,18 +112,22 @@ class CallbackThread : public wpi::SafeThread { template void CallbackThread::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::Main() { if (!listener) { continue; } + if (!static_cast(this)->Matches(listener, item.second)) { continue; } @@ -164,6 +175,11 @@ void CallbackThread::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 on_start) { + m_on_start = std::move(on_start); + } + + void SetOnExit(std::function 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 void DoStart(Args&&... args) { - m_owner.Start(std::forward(args)...); + m_owner.Start(m_on_start, m_on_exit, std::forward(args)...); } template @@ -361,6 +385,9 @@ class CallbackManager { private: wpi::SafeThreadOwner m_owner; + + std::function m_on_start; + std::function m_on_exit; }; } // namespace wpi