shmSubscriber.hpp

Go to the documentation of this file.

#ifndef RTCTK_STANDALONETOOLS_SHMSUBSCRIBER_H
#define RTCTK_STANDALONETOOLS_SHMSUBSCRIBER_H
 
#include <boost/io/ios_state.hpp>
#include <cassert>
#include <deque>
#include <iomanip>
#include <iostream>
#include <ipcq/adapter.hpp>
#include <ipcq/reader.hpp>
#include <limits>
#include <memory>
#include <string>
 
namespace rtctk::standaloneTools {

class ShmSubscriberBase {
public:
    ShmSubscriberBase() = default;
    // Do not allow copying or moving of this object.
    ShmSubscriberBase(const ShmSubscriberBase& rhs) = delete;
    ShmSubscriberBase& operator=(const ShmSubscriberBase& rhs) = delete;
    ShmSubscriberBase(ShmSubscriberBase&& rhs) = default;
    ShmSubscriberBase& operator=(ShmSubscriberBase&& rhs) = default;
 
    virtual ~ShmSubscriberBase() = default;
    int Run(int argc, char* argv[]);
 
protected:
    virtual void Initialise() = 0;

    virtual void Finalise() = 0;

    virtual bool ReadSample() = 0;

    virtual void PrintSample() = 0;

    virtual const void* GetSampleData() const = 0;

    virtual size_t GetSampleSize() const = 0;

    inline const std::string& GetQueueName() const {
        return m_queue_name;
    }

    inline const std::string& GetFilename() const {
        return m_filename;
    }

    inline const int64_t GetSampleNumber() const {
        return m_sample_counter;
    }

    inline const int64_t PrintWihtLongFormat() const {
        return m_print_long;
    }
 
private:
    bool ParseArguments(int argc, char* argv[]);
    void WriteBufferToFile(const void* buffer, size_t size);
    bool TerminateProcess();
 
    std::string m_queue_name;
    std::string m_filename;
    int64_t m_max_samples = 0;
    int64_t m_skip_samples = 0;
    bool m_print_samples = false;
    bool m_print_long = false;
    int64_t m_sample_counter = 0;
};

template <typename Topic,
          class ConditionPolicy = ipcq::BoostConditionPolicy,
          class ShmTraits = ipcq::detail::BoostInterprocessTraits>
class ShmSubscriber : public ShmSubscriberBase {
public:
    ShmSubscriber() = default;
    ~ShmSubscriber() override = default;
    // Do not allow copying or moving of this object.
    ShmSubscriber(const ShmSubscriber& rhs) = delete;
    ShmSubscriber& operator=(const ShmSubscriber& rhs) = delete;
    ShmSubscriber(ShmSubscriber&& rhs) noexcept = default;
    ShmSubscriber& operator=(ShmSubscriber&& rhs) noexcept = default;
 
protected:
    virtual void PrintSample(const Topic& sample) {
        boost::io::ios_flags_saver saved_state(std::cout);
        std::cout << "Sample " << GetSampleNumber() << ":\n";
        auto buffer = reinterpret_cast<const uint8_t*>(&sample);
        size_t max_bytes_to_print = 64;
        if (PrintWihtLongFormat()) {
            max_bytes_to_print = std::numeric_limits<size_t>::max();
        }
        bool last_was_endl = false;
        for (size_t n = 0; n < sizeof(Topic) and n < max_bytes_to_print; ++n) {
            std::cout << "0x" << std::setfill('0') << std::setw(2) << std::right << std::noshowbase
                      << std::hex << static_cast<unsigned int>(buffer[n]);
            if ((n + 1) % 16 == 0) {
                std::cout << "\n";
                last_was_endl = true;
            } else {
                std::cout << " ";
                last_was_endl = false;
            }
        }
        if (not last_was_endl) {
            std::cout << "\n";
        }
        if (sizeof(Topic) > max_bytes_to_print) {
            std::cout << "... (data continues) ...\n";
        }
    }
 
private:
    using Reader = ipcq::BasicReader<Topic, ConditionPolicy, ShmTraits>;

    void Initialise() override {
        try {
            m_reader = std::make_unique<Reader>(GetQueueName().c_str());
        } catch (const std::exception& error) {
            std::string msg = "Failed to create the shared memory reader for queue '" +
                              GetQueueName() + "': " + error.what();
            throw std::runtime_error(msg);
        }
    }

    void Finalise() override {
        try {
            m_reader.reset(nullptr);
        } catch (const std::exception& error) {
            std::string msg = "Failed to destroy the shared memory reader for queue '" +
                              GetQueueName() + "': " + error.what();
            throw std::runtime_error(msg);
        }
        m_samples.clear();
    }

    bool ReadSample() override {
        if (not m_samples.empty()) {
            m_samples.pop_front();
        }
        if (not m_samples.empty()) {
            return true;
        }
        using namespace std::chrono_literals;
        auto count = m_reader->NumAvailable();
        auto [error, num_elements] = m_reader->Read(ipcq::BackInserter(m_samples), count, 100ms);
        if (error) {
            if (error == ipcq::make_error_code(ipcq::Error::Timeout)) {
                return false;
            } else if (error == ipcq::make_error_code(ipcq::Error::InconsistentState)) {
                // Use case for this tool is not conserned about missed samples. So when we get
                // InconsistentState because of e.g. late joining we simply reset.
                // Reset may fail if queue is Closed or if it is empty (nothing to reset to). We
                // ignore that as well. Eventually there will be data in the queue and Reset() will
                // succeed, or queue will be closed and Reset won't be attempted again.
                if (!m_reader->Reset()) {
                    // Only log in the successful case as attempts to Reset will possibly otherwise
                    // flood the console with attempts if theres no data in the queue.
                    std::cerr << "Note: SHM reader state reset.\n";
                }
                return false;
            } else {
                std::string msg = "Failed to read from shared memory: " + error.message();
                throw std::runtime_error(msg);
            }
        }
        return num_elements > 0;
    }

    void PrintSample() override {
        assert(not m_samples.empty());
        PrintSample(m_samples.front());
    }

    const void* GetSampleData() const override {
        assert(not m_samples.empty());
        return reinterpret_cast<const void*>(&m_samples.front());
    }

    size_t GetSampleSize() const override {
        return sizeof(Topic);
    }
 
    std::deque<Topic> m_samples;
    std::unique_ptr<Reader> m_reader;
};
 
}  // namespace rtctk::standaloneTools
 
#endif  // RTCTK_STANDALONETOOLS_SHMSUBSCRIBER_H