Merge pull request #7 from ECFMP/event-bus

Event bus

Author: Andy Ford

include/flow-sdk/EventBus.h

@@ -0,0 +1,71 @@
+#pragma once
+#include "../src/eventbus/InternalEventStream.h"
+#include <any>
+#include <mutex>
+#include <typeindex>
+#include <unordered_map>
+
+namespace FlowSdk::EventBus {
+
+    template<typename EventType>
+    class EventStreamFactory;
+
+    class EventBus
+    {
+        public:
+        virtual ~EventBus() = default;
+        /**
+         * Subscribes the given listener to the event stream.
+         */
+        template<typename EventType>
+        void Subscribe(std::shared_ptr<NewEventListener<EventType>> listener)
+        {
+            Subscribe<EventType>(listener, nullptr);
+        };
+
+        template<typename EventType>
+        void Subscribe(
+                std::shared_ptr<NewEventListener<EventType>> listener, std::shared_ptr<NewEventFilter<EventType>> filter
+        )
+        {
+            GetStream<EventType>().Subscribe(listener, filter);
+        };
+
+        /**
+         * Subscribes the given listener to the event stream, but only for the next event.
+         */
+        template<typename EventType>
+        void SubscribeOnce(std::shared_ptr<NewEventListener<EventType>> listener)
+        {
+            SubscribeOnce<EventType>(listener, nullptr);
+        }
+
+        template<typename EventType>
+        void SubscribeOnce(
+                std::shared_ptr<NewEventListener<EventType>> listener, std::shared_ptr<NewEventFilter<EventType>> filter
+        )
+        {
+            GetStream<EventType>().SubscribeOnce(listener, filter);
+        };
+
+        private:
+        template<typename EventType>
+        auto GetStream() -> InternalEventStream<EventType>&
+        {
+            auto lock = std::lock_guard(mutex);
+
+            const auto index = std::type_index(typeid(EventType));
+            if (!streams.contains(index)) {
+                streams.insert({index, std::any(std::make_shared<InternalEventStream<EventType>>())});
+            }
+
+            return *std::any_cast<std::shared_ptr<InternalEventStream<EventType>>>(streams.at(index));
+        }
+
+        // Protects the streams map
+        std::mutex mutex;
+
+        // All the streams
+        std::unordered_map<std::type_index, std::any> streams;
+    };
+}// namespace FlowSdk::EventBus

include/flow-sdk/EventStream.h

@@ -0,0 +1,72 @@
+#pragma once
+#include <memory>
+#include <mutex>
+#include <vector>
+
+namespace FlowSdk::EventBus {
+
+    template<typename EventType>
+    class NewEventListener;
+    template<typename EventType>
+    class NewEventFilter;
+
+    template<typename EventType>
+    struct EventSubscription {
+        std::shared_ptr<NewEventListener<EventType>> listener;
+        std::shared_ptr<NewEventFilter<EventType>> filter;
+        bool once;
+    };
+
+    template<typename EventType>
+    class EventStream
+    {
+        public:
+        virtual ~EventStream() = default;
+
+        /**
+         * Subscribes the given listener to the event stream.
+         */
+        void Subscribe(std::shared_ptr<NewEventListener<EventType>> listener)
+        {
+            Subscribe(listener, nullptr);
+        };
+
+        virtual void Subscribe(
+                std::shared_ptr<NewEventListener<EventType>> listener, std::shared_ptr<NewEventFilter<EventType>> filter
+        )
+        {
+            if (listener == nullptr) {
+                throw std::invalid_argument("listener cannot be null");
+            }
+
+            auto guard = std::lock_guard(mutex);
+            subscriptions.push_back(EventSubscription<EventType>{listener, filter, false});
+        };
+
+        /**
+         * Subscribes the given listener to the event stream, but only for the next event.
+         */
+        virtual void SubscribeOnce(std::shared_ptr<NewEventListener<EventType>> listener)
+        {
+            SubscribeOnce(listener, nullptr);
+        }
+
+        virtual void SubscribeOnce(
+                std::shared_ptr<NewEventListener<EventType>> listener, std::shared_ptr<NewEventFilter<EventType>> filter
+        )
+        {
+            if (listener == nullptr) {
+                throw std::invalid_argument("listener cannot be null");
+            }
+
+            auto guard = std::lock_guard(mutex);
+            subscriptions.push_back(EventSubscription<EventType>{listener, filter, true});
+        };
+
+        protected:
+        std::mutex mutex;
+
+        // All subscriptions to this event stream.
+        std::vector<EventSubscription<EventType>> subscriptions;
+    };
+}// namespace FlowSdk::EventBus

include/flow-sdk/NewEventFilter.h

@@ -0,0 +1,12 @@
+#pragma once
+
+namespace FlowSdk::EventBus {
+
+    template<typename EventType>
+    class NewEventFilter
+    {
+        public:
+        virtual ~NewEventFilter() = default;
+        virtual bool ShouldProcess(const EventType&) = 0;
+    };
+}// namespace FlowSdk::EventBus

include/flow-sdk/NewEventListener.h

@@ -0,0 +1,11 @@
+#pragma once
+
+namespace FlowSdk::EventBus {
+    template<typename EventType>
+    class NewEventListener
+    {
+        public:
+        virtual ~NewEventListener() = default;
+        virtual void OnEvent(const EventType&) = 0;
+    };
+}// namespace FlowSdk::EventBus

include/flow-sdk/Sdk.h

@@ -1,13 +1,16 @@
 #pragma once
 
-namespace FlowSdk::FlowMeasure {
-    class FlowMeasure;
-}// namespace FlowSdk::FlowMeasure
+namespace FlowSdk {
+    namespace EventBus {
+        class EventBus;
+    }// namespace EventBus
+    namespace FlowMeasure {
+        class FlowMeasure;
+    }// namespace FlowMeasure
+}// namespace FlowSdk
 
 namespace FlowSdk::Plugin {
 
-    class SdkEventListeners;
-
     /**
      * Represents the public-facing facade of the SDK.
      */
@@ -16,7 +19,16 @@ namespace FlowSdk::Plugin {
         public:
         virtual ~Sdk() = default;
 
-        [[nodiscard]] virtual auto Listeners() const noexcept -> SdkEventListeners& = 0;
+        /**
+         * The event bus for the SDK, which can be used to subscribe to events.
+         */
+        [[nodiscard]] virtual auto EventBus() const noexcept -> EventBus::EventBus& = 0;
+
+        /**
+         * This method should be called once every second by the plugin using this SDK (as part of EuroScopes timer
+         * tick method).
+         */
+        virtual void OnEuroscopeTimerTick() = 0;
 
         /**
          * Destroys this SDK instance. Once destroyed it cannot be reused.

include/flow-sdk/SdkEventListenerTypes.h

@@ -0,0 +1,28 @@
+#pragma once
+
+namespace FlowSdk::FlowMeasure {
+    class FlowMeasure;
+}// namespace FlowSdk::FlowMeasure
+
+namespace FlowSdk::Plugin {
+    using FlowMeasureNotifiedEvent = struct FlowMeasureNotifiedEvent {
+        const FlowMeasure::FlowMeasure& measure;
+    };
+
+    using FlowMeasureActivatedEvent = struct FlowMeasureActivatedEvent {
+        const FlowMeasure::FlowMeasure& measure;
+    };
+
+    using FlowMeasureExpiredEvent = struct FlowMeasureExpiredEvent {
+        const FlowMeasure::FlowMeasure& measure;
+    };
+
+    using FlowMeasureWithdrawnEvent = struct FlowMeasureWithdrawnEvent {
+        const FlowMeasure::FlowMeasure& measure;
+    };
+
+    using FlowMeasureReissuedEvent = struct FlowMeasureReissuedEvent {
+        const FlowMeasure::FlowMeasure& original;
+        const FlowMeasure::FlowMeasure& reissued;
+    };
+}// namespace FlowSdk::Plugin

include/flow-sdk/SdkEventListeners.h

@@ -19,4 +19,4 @@
 #include "MultipleLevelFilter.h"
 #include "RouteFilter.h"
 #include "Sdk.h"
-#include "SdkEventListenerTypes.h"
+#include "SdkEvents.h"

include/flow-sdk/SdkEvents.h

@@ -14,7 +14,12 @@ set(src_flight_information_regions
 set(src_events
         event/ConcreteEvent.cpp event/ConcreteEvent.h
         event/ConcreteEventParticipant.cpp "event/ConcreteEventParticipant.h"
-        event/Event.cpp)
+        event/Event.cpp ../include/flow-sdk/EventStream.h ../include/flow-sdk/NewEventListener.h ../include/flow-sdk/NewEventFilter.h)
+
+set(src_eventbus
+        ../include/flow-sdk/EventStream.h
+        ../include/flow-sdk/NewEventListener.h
+        ../include/flow-sdk/NewEventFilter.h eventbus/InternalEventStream.h ../include/flow-sdk/EventBus.h eventbus/InternalEventBus.h)
 
 set(src_flowmeasures
         flowmeasure/ConcreteFlowMeasure.cpp flowmeasure/ConcreteFlowMeasure.h flowmeasure/ConcreteAirportFilter.cpp flowmeasure/ConcreteAirportFilter.h flowmeasure/ConcreteEventFilter.cpp flowmeasure/ConcreteEventFilter.h flowmeasure/ConcreteLevelRangeFilter.cpp flowmeasure/ConcreteLevelRangeFilter.h flowmeasure/ConcreteRouteFilter.cpp flowmeasure/ConcreteRouteFilter.h flowmeasure/ConcreteMeasure.cpp flowmeasure/ConcreteMeasure.h flowmeasure/ConcreteMeasureFactory.cpp flowmeasure/ConcreteMeasureFactory.h flowmeasure/ConcreteFlowMeasureFilters.cpp flowmeasure/ConcreteFlowMeasureFilters.h api/FlowMeasureFilterParserInterface.h ../include/flow-sdk/MultipleLevelFilter.h flowmeasure/ConcreteMultipleLevelFilter.cpp flowmeasure/ConcreteMultipleLevelFilter.h ../include/mock/MultipleLevelFilterMock.h ../include/flow-sdk/RangeToDestinationFilter.h flowmeasure/ConcreteRangeToDestinationFilter.cpp flowmeasure/ConcreteRangeToDestinationFilter.h)
@@ -28,12 +33,13 @@ set(src_plugin
         plugin/ConcreteSdk.h
         plugin/ConcreteSdk.cpp
         plugin/InternalEventListeners.h
-        plugin/ConcreteEventListeners.cpp plugin/ConcreteEventListeners.h plugin/ConcreteSdkEventListeners.cpp plugin/ConcreteSdkEventListeners.h plugin/SdkFactory.cpp)
+        plugin/ConcreteEventListeners.cpp plugin/ConcreteEventListeners.h plugin/SdkFactory.cpp ../include/flow-sdk/SdkEvents.h)
 
 set(ALL_FILES
         ${src_api}
         ${src_date}
         ${src_events}
+        ${src_eventbus}
         ${src_flowmeasures}
         ${src_flight_information_regions}
         ${src_log}