ReadStream.h

Go to the documentation of this file.

#pragma once
 
#include <functional>
#include <optional>
#include <thread>
 
#include <ArmarXCore/core/time/DateTime.h>
#include <ArmarXCore/core/time/Frequency.h>
#include <ArmarXCore/core/time/Metronome.h>
 
#include <RobotAPI/libraries/armem/core/MemoryID.h>
#include <RobotAPI/libraries/armem/core/wm/memory_definitions.h>
 
#include "Reader.h"
 
namespace armarx::armem::client
{
    /**
     * @brief A stream reading entity snapshots from the memory.
     *
     * After constructing a ReadStream, polling can be started in three ways:
     *
     * 1. Run a polling loop in this thread, blocking execution until terminated.
     *    See pollBlocking().
     * 2. Run a polling loop in a new, separate thread, until it is stopped via stop().
     *    See pollAsync() and stop().
     * 3. Perform a single query and process the result, embedded in your own loop or other control
     *    flow logic.
     *    See pollOnce().
     */
    class ReadStream
    {
    public:
        /**
         * @brief Callback called on each entity snapshot in the queried ID.
         *
         * If it returns false, the stream is stopped.
         */
        using SnapshotCallbackT = std::function<bool(const wm::EntitySnapshot&)>;
 
        /**
         * @brief Inizialize a ReadStream which does not represent a stream.
         */
        ReadStream();
 
        /**
         * @brief Initialize a read stream.
         *
         * @param reader
         *   The reader to perform the queries.
         * @param queriedId
         *   The memory ID in which all snapshots should be processed by the stream.
         * @param maxPollFrequency
         *   The maximum frequency with which queries are performed. The
         *   real frequency might be lower.
         */
        ReadStream(const Reader& reader,
                   const MemoryID& queriedId,
                   const armarx::core::time::Frequency& maxPollFrequency =
                       armarx::core::time::Frequency::Hertz(10));
 
 
        /**
         * @brief Poll in this thread as long as callback returns true.
         *
         * @param callback Function to call on each entity snapshot.
         * @return The snapshot object that returns false.
         * @throw armarx::armem::error::ReadStreamAlreadyPolling If the stream is already polling.
         */
        std::optional<wm::EntitySnapshot> pollBlocking(const SnapshotCallbackT& callback);
 
        /**
         * @brief Poll in a new thread as long as callback returns true.
         *
         * Note that callback will be called in a separate thread, so take care of synchronizing
         * access to variables in the callback appropriately.
         *
         * Roughly equivalent to:
         * @code
         * std::thread thread([]() { stream.pollBlocking(); });
         * @endcode
         *
         * @param callback Function to call on each entity snapshot.
         * @throw armarx::armem::error::ReadStreamAlreadyPolling If the stream is already polling.
         */
        void pollAsync(const SnapshotCallbackT& callback);
        /**
         * @brief Stop a running polling loop.
         *
         * If a polling thread has been started by pollAsync() before, joins the thread.
         */
        void stop();
 
        /**
         * @brief Perform one query and call the callbacks on each snapshot.
         *
         * This allows you to define your own loop, for example:
         *
         * @code
         * bool condition = true;
         * while (condition)
         * {
         *     auto snapshot = stream.pollOnce(callback);
         *
         *     ...
         *
         *     if (...)
         *     {
         *        condition = false;
         *     }
         * }
         * @endcode
         *
         * @param callback Function to call on each entity snapshot.
         * @throw armarx::armem::error::ReadStreamAlreadyPolling If the stream is already polling.
         */
        std::optional<wm::EntitySnapshot> pollOnce(const SnapshotCallbackT& callback);
 
 
    private:
        std::optional<wm::EntitySnapshot> _pollBlocking(const SnapshotCallbackT& callback);
        std::optional<wm::EntitySnapshot> _pollOnce(const SnapshotCallbackT& callback);
 
 
    private:
        Reader reader;
        MemoryID queriedId;
 
        armarx::core::time::Metronome metronome;
        armarx::DateTime timeStart;
 
        std::atomic_bool isPolling = false;
        std::atomic_bool pollingStoppedExternally = false;
 
        std::thread pollingThread;
    };
 
} // namespace armarx::armem::client