lasp_streammgr.h

Go to the documentation of this file.

#pragma once
#include "lasp_daq.h"
#include "lasp_siggen.h"
#include "lasp_thread.h"
#include <list>
#include <memory>
#include <mutex>
#include <thread>
 
class StreamMgr;
class InDataHandler;
 
 
class SeriesBiquad;
 
class StreamMgr {
 
  std::unique_ptr<Daq> _inputStream, _outputStream;
 
  GlobalThreadPool _pool;
 
  std::list<InDataHandler *> _inDataHandlers;
  mutable std::mutex _inDataHandler_mtx;
 
  std::shared_ptr<Siggen> _siggen;
  std::mutex _siggen_mtx;
 
  std::vector<std::unique_ptr<SeriesBiquad>> _inputFilters;
 
 
  mutable std::recursive_mutex _devices_mtx;
  DeviceInfoList _devices;
 
  // Singleton, no public constructor. Can only be obtained using
  // getInstance();
  StreamMgr();
 
  friend class InDataHandler;
  friend class Siggen;
 
 
  public:
 
  ~StreamMgr();
 
  enum class StreamType : us {
    input = 1 << 0,
    output = 1 << 1,
  };
 
  StreamMgr(const StreamMgr &) = delete;
  StreamMgr &operator=(const StreamMgr &) = delete;
 
  static std::shared_ptr<StreamMgr> getInstance();
 
  DeviceInfoList getDeviceInfo() const {
  std::scoped_lock lck(_devices_mtx);
    DeviceInfoList d2;
    for(const auto& dev: _devices) {
      d2.push_back(dev->clone());
    }
    return d2;
  }
 
  void
  rescanDAQDevices(bool background = false,
                   std::function<void()> callback = std::function<void()>());
 
  void startStream(const DaqConfiguration &config);
 
  bool isStreamRunningOK(const StreamType type) const {
    return getStreamStatus(type).runningOK();
  }
  bool isStreamRunning(const StreamType type) const {
    switch (type) {
    case (StreamType::input):
      return bool(_inputStream);
      break;
    case (StreamType::output):
      return bool(_outputStream);
      break;
    }
    return false;
  }
 
  Daq::StreamStatus getStreamStatus(const StreamType type) const;
 
  const Daq *getDaq(StreamType type) const;
 
  void stopStream(const StreamType stype);
 
  void stopAllStreams();
 
  void setSiggen(std::shared_ptr<Siggen> s);
 
private:
  void inCallback(const DaqData &data);
  void outCallback(DaqData &data);
 
 
  void addInDataHandler(InDataHandler *handler);
 
  void removeInDataHandler(InDataHandler &handler);
  void rescanDAQDevices_impl(std::function<void()> callback);
 
#if LASP_DEBUG == 1
  const std::thread::id main_thread_id;
  void checkRightThread() const;
#else
  void checkRightThread() const {}
#endif
};