From ed4b80d1f5df4db016a7178c650f39d7da70144f Mon Sep 17 00:00:00 2001 From: Dawid Drozd Date: Sat, 17 Jun 2023 15:01:52 +0400 Subject: [PATCH] Add Listener::isListening Requested feature in #48 --- lib/CMakeLists.txt | 2 +- lib/src/dexode/EventBus.cpp | 12 +++++++ lib/src/dexode/EventBus.hpp | 3 ++ lib/src/dexode/eventbus/Bus.hpp | 3 ++ lib/src/dexode/eventbus/Listener.hpp | 13 +++++++ .../eventbus/internal/ListenerAttorney.hpp | 13 +++++++ .../dexode/eventbus/stream/EventStream.hpp | 6 ++++ .../eventbus/stream/ProtectedEventStream.hpp | 11 ++++-- .../dexode/eventbus/test/SuiteListener.cpp | 34 +++++++++++++++++++ 9 files changed, 94 insertions(+), 3 deletions(-) diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt index 0755584..3729263 100644 --- a/lib/CMakeLists.txt +++ b/lib/CMakeLists.txt @@ -5,7 +5,7 @@ set(CPACK_GENERATOR "" CACHE STRING "Set packages CPack should build e.g. ZIP;TG # BUILD_SHARED_LIBS can controll build type! project(EventBus - VERSION 3.0.3 + VERSION 3.0.4 LANGUAGES CXX ) diff --git a/lib/src/dexode/EventBus.cpp b/lib/src/dexode/EventBus.cpp index 19c37af..b388e3a 100644 --- a/lib/src/dexode/EventBus.cpp +++ b/lib/src/dexode/EventBus.cpp @@ -104,6 +104,18 @@ eventbus::stream::EventStream* EventBus::obtainStream( } } +eventbus::stream::EventStream* EventBus::streamForEvent( + eventbus::internal::event_id_t eventID) const +{ + std::lock_guard writeGuard{_mutexStreams}; + auto* found = findStreamUnsafe(eventID); + if(found != nullptr) + { + return found; + } + return nullptr; +} + bool EventBus::postponeEvent(eventbus::PostponeHelper& postponeCall) { auto* eventStream = obtainStream(postponeCall.eventID, postponeCall.createStreamCallback); diff --git a/lib/src/dexode/EventBus.hpp b/lib/src/dexode/EventBus.hpp index 2c9ee2d..50c139c 100644 --- a/lib/src/dexode/EventBus.hpp +++ b/lib/src/dexode/EventBus.hpp @@ -29,6 +29,9 @@ public: std::size_t processLimit(std::size_t limit); protected: + eventbus::stream::EventStream* streamForEvent( + eventbus::internal::event_id_t eventID) const override; + eventbus::stream::EventStream* obtainStream( eventbus::internal::event_id_t eventID, eventbus::CreateStreamCallback createStreamCallback); diff --git a/lib/src/dexode/eventbus/Bus.hpp b/lib/src/dexode/eventbus/Bus.hpp index d72d3e1..a0bb01f 100644 --- a/lib/src/dexode/eventbus/Bus.hpp +++ b/lib/src/dexode/eventbus/Bus.hpp @@ -92,6 +92,9 @@ protected: virtual void unlistenAll(std::uint32_t listenerID) = 0; virtual void unlisten(std::uint32_t listenerID, internal::event_id_t eventID) = 0; + virtual eventbus::stream::EventStream* streamForEvent( + eventbus::internal::event_id_t eventID) const = 0; + private: std::atomic _lastID{0}; diff --git a/lib/src/dexode/eventbus/Listener.hpp b/lib/src/dexode/eventbus/Listener.hpp index 8e298ab..525a97d 100644 --- a/lib/src/dexode/eventbus/Listener.hpp +++ b/lib/src/dexode/eventbus/Listener.hpp @@ -131,6 +131,19 @@ public: return _bus; } + template + [[nodiscard]] bool isListening() const + { + static_assert(internal::validateEvent(), "Invalid event"); + if(_bus == nullptr) + { + throw std::runtime_error{"bus is null"}; + } + return internal::ListenerAttorney::isListening(*_bus + , _id + , internal::event_id()); + } + private: std::uint32_t _id = 0; std::shared_ptr _bus = nullptr; diff --git a/lib/src/dexode/eventbus/internal/ListenerAttorney.hpp b/lib/src/dexode/eventbus/internal/ListenerAttorney.hpp index 1fc2d74..304db1c 100644 --- a/lib/src/dexode/eventbus/internal/ListenerAttorney.hpp +++ b/lib/src/dexode/eventbus/internal/ListenerAttorney.hpp @@ -6,6 +6,7 @@ #include #include "dexode/eventbus/internal/event_id.hpp" +#include "dexode/eventbus/stream/EventStream.hpp" namespace dexode::eventbus { @@ -49,6 +50,18 @@ private: { bus.unlisten(listenerID, eventID); } + + static constexpr bool isListening(EventBus_t& bus, + const std::uint32_t listenerID, + const event_id_t eventID) + { + const eventbus::stream::EventStream* stream = bus.streamForEvent(eventID); + if(stream != nullptr) + { + return stream->hasListener(listenerID); + } + return false; + } }; } // namespace dexode::eventbus::internal diff --git a/lib/src/dexode/eventbus/stream/EventStream.hpp b/lib/src/dexode/eventbus/stream/EventStream.hpp index e85b4e2..a40f8f5 100644 --- a/lib/src/dexode/eventbus/stream/EventStream.hpp +++ b/lib/src/dexode/eventbus/stream/EventStream.hpp @@ -16,6 +16,8 @@ public: virtual bool addListener(std::uint32_t listenerID, std::any callback) = 0; virtual bool removeListener(std::uint32_t listenerID) = 0; + + [[nodiscard]] virtual bool hasListener(std::uint32_t listenerID) const = 0; }; class NoopEventStream : public EventStream @@ -37,6 +39,10 @@ public: { throw std::runtime_error{"Noop"}; } + [[nodiscard]] bool hasListener(std::uint32_t listenerID) const override + { + throw std::runtime_error{"Noop"}; + } }; } // namespace dexode::eventbus::stream diff --git a/lib/src/dexode/eventbus/stream/ProtectedEventStream.hpp b/lib/src/dexode/eventbus/stream/ProtectedEventStream.hpp index 40d574b..ad60896 100644 --- a/lib/src/dexode/eventbus/stream/ProtectedEventStream.hpp +++ b/lib/src/dexode/eventbus/stream/ProtectedEventStream.hpp @@ -98,6 +98,13 @@ public: return not _queue.empty(); } + [[nodiscard]] bool hasListener(std::uint32_t listenerID) const override + { + std::shared_lock readGuard{_mutexCallbacks}; + auto found = std::find(_listenerIDs.begin(), _listenerIDs.end(), listenerID); + return found != _listenerIDs.end(); + } + private: std::vector _listenerIDs; std::vector _queue; @@ -106,8 +113,8 @@ private: std::atomic _isProcessing{false}; std::vector> _waiting; - std::shared_mutex _mutexEvent; - std::shared_mutex _mutexCallbacks; + mutable std::shared_mutex _mutexEvent; + mutable std::shared_mutex _mutexCallbacks; void flushWaitingOnes() { diff --git a/test/unit/src/dexode/eventbus/test/SuiteListener.cpp b/test/unit/src/dexode/eventbus/test/SuiteListener.cpp index deb7afa..64a1e3c 100644 --- a/test/unit/src/dexode/eventbus/test/SuiteListener.cpp +++ b/test/unit/src/dexode/eventbus/test/SuiteListener.cpp @@ -297,4 +297,38 @@ TEST_CASE("Should not allow for mistake with move ctor", "[EventBus][Listener]") REQUIRE(TestClazz::counter == 1); } + +TEST_CASE("Should allow to check if listener is already listening", "[EventBus][Listener]") +{ + // Related to Github Issue: https://github.com/gelldur/EventBus/issues/48 + EventBus bus; + int callCount = 0; + auto listener = Listener::createNotOwning(bus); + + CHECK_FALSE(listener.isListening()); + + bus.postpone(event::Value{3}); + REQUIRE(bus.process() == 1); + REQUIRE(callCount == 0); // not listening + + listener.listen([&](const event::Value& event) + { + REQUIRE(event.value == 2); + ++callCount; + }); + CHECK(listener.isListening()); + + bus.postpone(event::Value{2}); + REQUIRE(bus.process() == 1); + REQUIRE(callCount == 1); + + CHECK(listener.isListening()); + listener.unlisten(); + CHECK_FALSE(listener.isListening()); + + bus.postpone(event::Value{1}); + REQUIRE(bus.process() == 1); + REQUIRE(callCount == 1); +} + } // namespace dexode::eventbus::test