misc/InteractiveStreamingExample/JWindow.h

0001
0002 // Copyright 2020, Jefferson Science Associates, LLC.
0003 // Subject to the terms in the LICENSE file found in the top-level directory.
0004
0005 #pragma once
0006 #include "JMessage.h"
0007 #include <JANA/JEvent.h>
0008
0009 #include <queue>
0010
0011 /// JWindow is an abstract data structure for aggregating individual JMessages into a
0012 /// single JEvent.  We generally assume that messages from any particular source arrive in-order, and
0013 /// make no assumptions about ordering between different sources. We provide different implementations to fit
0014 /// different use cases so that the user should rarely need to implement one of these by themselves.
0015 /// The choice of JWindow determines how the time interval associated with each
0016 /// JEvent is calculated, and furthermore is responsible for placing the correct JMessages into
0017 /// the JEvent. As long as the time intervals do not overlap, this amounts to simple transferring
0018 /// of ownership of a raw pointer. When JEvents intervals do overlap, which happens in the case of JSlidingWindow
0019 /// and JMergeWindow, we have to decide whether to use shared ownership or to clone the offending data.
0020 template <typename T>
0021 struct JWindow {
0022
0023     virtual ~JWindow() = default;
0024     virtual void pushMessage(T* message) = 0;
0025     virtual bool pullEvent(JEvent& event) = 0;
0026
0027 };
0028
0029
0030 /// first JMessage and ends once there are no more JMessages timestamped before a configurable
0031 /// max interval width. This is usually what is meant by 'event-building'.
0032 template <typename T>
0033 class JSessionWindow : public JWindow<T> {
0034 private:
0035     template <typename S>
0036     struct maybe {
0037         bool is_none = true;
0038         S value;
0039     };
0040
0041 public:
0042
0043     JSessionWindow(Timestamp event_interval, const std::vector<DetectorId> &detectors)
0044             : m_event_interval(event_interval) {
0045         for (auto id : detectors) {
0046             m_inbox.insert({id, {}});
0047         }
0048     }
0049
0050     void pushMessage(T* message) final {
0051         // TODO: Why was messages a vector before?
0052         auto iter = m_inbox.find(message->source_id);
0053         if (iter == m_inbox.end()) {
0054             throw std::runtime_error("Unexpected detector!");
0055         } else {
0056             iter->second.push_back(message);
0057         }
0058     };
0059
0060     bool pullEvent(JEvent&) final {
0061         return false;
0062     };
0063
0064 /*
0065     maybe<JMessage::Timestamp> find_next_event_start() {
0066         maybe<JMessage::Timestamp> result;
0067         result.is_none = true;
0068         if (!m_inbox.empty()) {
0069             result.is_none = false;
0070             auto iter = m_inbox.begin();
0071             result.value = iter->second.timestamp;
0072
0073             while (++iter != m_inbox.end()) {
0074                 if (iter->second.is_none) {
0075                     result.is_none = true;
0076                     return result;
0077                 }
0078                 result.value = std::min(result.value, iter->second.value);
0079             }
0080         }
0081         return result;
0082     }
0083
0084     bool stage_next_event() {
0085         if (m_next_event_start.is_none) {
0086             m_next_event_start = find_next_event_start();
0087             if (m_next_event_start.is_none) {
0088                 return false;
0089             }
0090         }
0091         Timestamp next_event_finish = m_next_event_start.value + m_event_interval;
0092         for (auto iter : m_inbox) { // for each detector
0093             auto &q = iter.second;
0094             while (q.front().timestamp <= next_event_finish) {
0095                 m_outbox.push_back(std::move(q.front()));
0096                 q.pop_front();
0097             }
0098             if (q.empty()) {
0099                 // we've run out of samples, so we can't say whether we have everything or not
0100                 return false;
0101             }
0102         }
0103         return true;
0104     }
0105
0106 */
0107 private:
0108     std::map<DetectorId, std::deque<JMessage*>> m_inbox;
0109     std::vector<T *> m_outbox;
0110     maybe<Timestamp> m_next_event_start;
0111     Timestamp m_event_interval; // TODO: This should be a duration
0112 };
0113