decl_composite_stream_base.hpp

#pragma once
 
#include <memory>
#include <atomic>
 
#include "../stream_base.hpp"
#include "../driver.hpp"
#include "../../exception.hpp"
#include "../../cancellation_token.hpp"
#include "../../genapi/node_map_enumerator.hpp"
#include "../flow_set_pool.hpp"
#include "../gendc_descriptor.hpp"
 
namespace Cvb
{
 
  CVB_BEGIN_INLINE_NS
 
  template <>
  inline HandleGuard<CompositeStreamBase>::HandleGuard(void *handle) noexcept
      : HandleGuard<CompositeStreamBase>(handle, [](void *handle) { CVB_CALL_CAPI(ReleaseObject(handle)); })
  {
  }
 
  namespace Driver
  {


    class CompositeStreamBase : public StreamBase
    {
    public:
      using GuardType = HandleGuard<CompositeStreamBase>;
 
      template <class T, class... ARGS>
      static std::shared_ptr<T> FromHandle(HandleGuard<CompositeStreamBase> &&guard, ARGS &&...args);


      void *Handle() const noexcept
      {
        return handle_.Handle();
      }
      void *Handle() const noexcept {…}


      virtual void EngineStart()
      {
        EngineStart(-1);
      }
      virtual void EngineStart() {…}


      virtual void EngineStart(std::int64_t bufferCountToAcquire)
      {
        Internal::DoResCall([this, bufferCountToAcquire]() {
          return CExports::CVDStreamAcquisitionEngineStart(Handle(), bufferCountToAcquire);
        });
        acquisitionState_ = Cvb::AcquisitionState::Started;
      }
      virtual void EngineStart(std::int64_t bufferCountToAcquire) {…}


      template <class Rep, class Period>
      void EngineStop(const std::chrono::duration<Rep, Period> &timeout)
      {
        InternalEngineStop(std::chrono::duration_cast<std::chrono::milliseconds>(timeout));
      }
      void EngineStop(const std::chrono::duration<Rep, Period> &timeout) {…}


      virtual void EngineStop()
      {
        InternalEngineStop(std::chrono::milliseconds(-1));
      }
      virtual void EngineStop() {…}


      template <class Rep, class Period>
      bool TryEngineStop(const std::chrono::duration<Rep, Period> &timeout) noexcept
      {
        return TryEngineStop(std::chrono::duration_cast<std::chrono::milliseconds>(timeout));
      }
      bool TryEngineStop(const std::chrono::duration<Rep, Period> &timeout) noexcept {…}


      virtual bool TryEngineStop() noexcept
      {
        return TryEngineStop(std::chrono::milliseconds(-1));
      }
      virtual bool TryEngineStop() noexcept {…}


      virtual bool TryEngineStop(const std::chrono::milliseconds &timeout) noexcept
      {
        if (acquisitionState_ == Cvb::AcquisitionState::Started)
          acquisitionState_ = Cvb::AcquisitionState::Stopping;
        auto result = CExports::CVDStreamAcquisitionEngineStop(Handle(), timeout.count()) == ErrorCodes::CVB_OK;
        if (result)
          acquisitionState_ = Cvb::AcquisitionState::Stopped;
        return result;
      }
      virtual bool TryEngineStop(const std::chrono::milliseconds &timeout) noexcept {…}


      virtual void EngineAbort()
      {
        if (acquisitionState_ == Cvb::AcquisitionState::Started)
          acquisitionState_ = Cvb::AcquisitionState::Stopping;
        else if (acquisitionState_ == Cvb::AcquisitionState::Stopping)
          acquisitionState_ = Cvb::AcquisitionState::AbortingStop;
        Internal::DoResCall([this]() { return CExports::CVDStreamAcquisitionEngineAbort(Handle()); });
        acquisitionState_ = Cvb::AcquisitionState::Stopped;
      }
      virtual void EngineAbort() {…}


      virtual bool TryEngineAbort() noexcept
      {
        if (acquisitionState_ == Cvb::AcquisitionState::Started)
          acquisitionState_ = Cvb::AcquisitionState::Stopping;
        else if (acquisitionState_ == Cvb::AcquisitionState::Stopping)
          acquisitionState_ = Cvb::AcquisitionState::AbortingStop;
        auto result = CExports::CVDStreamAcquisitionEngineAbort(Handle()) == ErrorCodes::CVB_OK;
        if (result)
          acquisitionState_ = Cvb::AcquisitionState::Stopped;
        return result;
      }
      virtual bool TryEngineAbort() noexcept {…}


      virtual void DeviceStart()
      {
        DeviceStart(-1);
      }
      virtual void DeviceStart() {…}


      virtual void DeviceStart(std::int64_t bufferCountToAcquire)
      {
        Internal::DoResCall(
            [this, bufferCountToAcquire]() { return CExports::CVDStreamControlStart(Handle(), bufferCountToAcquire); });
      }
      virtual void DeviceStart(std::int64_t bufferCountToAcquire) {…}


      virtual void DeviceStop()
      {
        Internal::DoResCall([this]() { return CExports::CVDStreamControlStop(Handle()); });
      }
      virtual void DeviceStop() {…}


      virtual bool TryDeviceStop() noexcept
      {
        return CExports::CVDStreamControlStop(Handle()) == ErrorCodes::CVB_OK;
      }
      virtual bool TryDeviceStop() noexcept {…}


      virtual void DeviceAbort()
      {
        Internal::DoResCall([this]() { return CExports::CVDStreamControlAbort(Handle()); });
      }
      virtual void DeviceAbort() {…}


      virtual bool TryDeviceAbort() noexcept
      {
        return CExports::CVDStreamControlAbort(Handle()) == ErrorCodes::CVB_OK;
      }
      virtual bool TryDeviceAbort() noexcept {…}


      void DeregisterFlowSetPool()
      {
        CVB_CALL_CAPI_CHECKED(CVDStreamUnregisterFlowSetPool(Handle()));
      }
      void DeregisterFlowSetPool() {…}


      void RegisterManagedFlowSetPool(int flowSetCount)
      {
        CVB_CALL_CAPI_CHECKED(CVDStreamRegisterManagedFlowSetPool(Handle(), static_cast<std::size_t>(flowSetCount)));
      }
      void RegisterManagedFlowSetPool(int flowSetCount) {…}


      void RegisterExternalFlowSetPool(FlowSetPoolPtr flowSetPoolPtr)
      {
        RegisterExternalFlowSetPool(flowSetPoolPtr, []() {});
      }
      void RegisterExternalFlowSetPool(FlowSetPoolPtr flowSetPoolPtr) {…}


      void RegisterExternalFlowSetPool(FlowSetPoolPtr flowSetPoolPtr, std::function<void()> releaseCallback)
      {
        flowSetPoolPtr->SetUserCallback(releaseCallback);
        CVB_CALL_CAPI_CHECKED(CVDStreamRegisterExternalFlowSetPool(
            Handle(), reinterpret_cast<CExports::CVDFlowSet *>(flowSetPoolPtr->ToNativeStruct().data()),
            flowSetPoolPtr->PoolSize(), flowSetPoolPtr->OnReleaseCallback, &*flowSetPoolPtr));
        flowSetPoolPtr->SetReleaseThis(flowSetPoolPtr);
      }
      void RegisterExternalFlowSetPool(FlowSetPoolPtr flowSetPoolPtr, std::function<void()> releaseCallback) {…}


      int FlowSetCount() const noexcept
      {
        std::size_t count = 0;
        CVB_CALL_CAPI(CVDStreamGetFlowSetCount(Handle(), count));
        return static_cast<int>(count);
      }
      int FlowSetCount() const noexcept {…}


      int MinRequiredFlowSetCount() const
      {
        std::size_t count = 0;
        CVB_CALL_CAPI(CVDStreamFlowSetInfoGetFlowSetMinCount(Handle(), count));
        return static_cast<int>(count);
      }
      int MinRequiredFlowSetCount() const {…}
 
      Cvb::Driver::AcquisitionInterface AcquisitionInterface() const noexcept override
      {
        return Cvb::Driver::AcquisitionInterface::GenTL;
      }
      Cvb::Driver::AcquisitionInterface AcquisitionInterface() const noexcept override {…}
 
      void Start() override
      {
        EngineStart();
        DeviceStart();
        combinedStreaming_ = true;
      }
      void Start() override {…}
 
      void Stop() override
      {
        DeviceStop();
        EngineStop();
        combinedStreaming_ = false;
      }
      void Stop() override {…}
 
      bool TryStop() noexcept override
      {
        auto deviceResult  = TryDeviceStop();
        auto engineResult  = TryEngineStop();
        combinedStreaming_ = false;
        return deviceResult && engineResult;
      }
      bool TryStop() noexcept override {…}
 
      void Abort() override
      {
        DeviceAbort();
        EngineAbort();
        combinedStreaming_ = false;
      }
      void Abort() override {…}
 
      bool TryAbort() noexcept override
      {
        auto deviceResult  = TryDeviceAbort();
        auto engineResult  = TryEngineAbort();
        combinedStreaming_ = false;
        return deviceResult && engineResult;
      }
      bool TryAbort() noexcept override {…}
 
      bool IsIndexed() const noexcept override
      {
        return false;
      }
      bool IsIndexed() const noexcept override {…}


      Cvb::AcquisitionState AcquisitionState() const noexcept
      {
        return acquisitionState_;
      }
      Cvb::AcquisitionState AcquisitionState() const noexcept {…}


      std::unique_ptr<Driver::GenDcDescriptor> GenDCPrefetchDescriptor() const
      {
        if (!CExports::CVDCanGenDCDescriptorProvider(Handle()))
          throw std::runtime_error("Stream does not represent a GenDC stream");
 
        return Internal::DoResCallObjectOut<Driver::GenDcDescriptor>([this](CExports::CVDGNDCDESCRIPTOR &handle) {
          return CVB_CALL_CAPI(CVDGNDCPGetDescriptor(Handle(), handle));
        });
      }
      std::unique_ptr<Driver::GenDcDescriptor> GenDCPrefetchDescriptor() const {…}


      std::vector<FlowInfo> FlowSetInfo() const
      {
        if (!CExports::CVDCanFlowSetInfo(Handle()))
          throw std::runtime_error("Stream has no flow set info");
 
        // Get the flow count
        std::size_t flowCount = 0;
        CVB_CALL_CAPI_CHECKED(CVDStreamFlowSetInfoGetFlowCount(Handle(), flowCount));
 
        std::vector<FlowInfo> flowInfos(flowCount);
        for (std::size_t i = 0; i < flowCount; ++i)
        {
          CVB_CALL_CAPI_CHECKED(CVDStreamFlowSetInfoGetFlowSize(Handle(), i, flowInfos[i].Size));
          CVB_CALL_CAPI_CHECKED(CVDStreamFlowSetInfoGetFlowAlignment(Handle(), i, flowInfos[i].Alignment));
        }
 
        return flowInfos;
      }
      std::vector<FlowInfo> FlowSetInfo() const {…}


      NodeMapPtr NodeMap(const String &name) const;


      std::map<String, NodeMapPtr> NodeMaps() const;
 
      CompositeStreamBase(const CompositeStreamBase &other)            = delete;
      CompositeStreamBase &operator=(const CompositeStreamBase &other) = delete;
      CompositeStreamBase(CompositeStreamBase &&other)                 = delete;
      CompositeStreamBase &operator=(CompositeStreamBase &&other)      = delete;
      virtual ~CompositeStreamBase()
      {
        if (combinedStreaming_)
        {
          // duplicate TryAbort() avoid virtual dispatch bypass
          CExports::CVDStreamControlAbort(Handle());
          CExports::CVDStreamAcquisitionEngineAbort(Handle());
        }
      }
 
    protected:
      template <class T>
      using WaitResultBase = std::tuple<HandleGuard<T>, WaitStatus>;
 
      explicit CompositeStreamBase(const DevicePtr &device, HandleGuard<CompositeStreamBase> &&guard)
          : StreamBase(device)
          , handle_(std::move(guard))
          , acquisitionState_(Cvb::AcquisitionState::Stopped)
          , combinedStreaming_(false)
      {
        if (CExports::CanNodeMapHandle2(Handle()))
          TryFillNodeMapKeys2();
      }
 
      template <class T, class Rep, class Period>
      WaitResultBase<T> InternalWaitFor(const std::chrono::duration<Rep, Period> &timeSpan)
      {
        return InternalWaitFor<T>(std::chrono::duration_cast<std::chrono::milliseconds>(timeSpan).count());
      }
 
      template <class T, class Rep, class Period>
      WaitResultBase<T> InternalWaitFor(const std::chrono::duration<Rep, Period> &timeSpan,
                                        const CancellationToken &token)
      {
        return InternalWaitFor<T>(std::chrono::duration_cast<std::chrono::milliseconds>(timeSpan).count(),
                                  token.Handle());
      }
 
      template <class T>
      WaitResultBase<T> InternalWait()
      {
        return InternalWaitFor<T>(-1);
      }
 
      template <class T>
      WaitResultBase<T> InternalWait(const CancellationToken &token)
      {
        return InternalWaitFor<T>(-1, token.Handle());
      }
 
    private:
      template <class T>
      WaitResultBase<T> InternalWaitFor(std::int64_t timeout, CExports::CVCANCELLATIONTOKEN token = nullptr)
      {
        CExports::CVDWaitStatus waitStatus = CExports::CVDWaitStatus::CVDWS_Ok;
        CExports::CVCOMPOSITE handle       = nullptr;
        CVB_CALL_CAPI_CHECKED(
            CVDStreamAcquisitionEngineWaitForNextComposite(Handle(), timeout, token, waitStatus, handle));
        return std::make_tuple(HandleGuard<T>(handle), static_cast<WaitStatus>(waitStatus));
      }
 
      virtual void InternalEngineStop(const std::chrono::milliseconds &timeout)
      {
        if (acquisitionState_ == Cvb::AcquisitionState::Started)
          acquisitionState_ = Cvb::AcquisitionState::Stopping;
        Internal::DoResCall(
            [this, timeout]() { return CExports::CVDStreamAcquisitionEngineStop(Handle(), timeout.count()); });
        acquisitionState_ = Cvb::AcquisitionState::Stopped;
      }
 
      bool TryFillNodeMapKeys2()
      {
        size_t numNodeMaps = 0;
        auto resultNum     = CExports::NMH2GetNum(Handle(), numNodeMaps);
        if (resultNum < 0)
          return false;
        for (size_t i = 0; i < numNodeMaps; ++i)
        {
          size_t bufferSize       = 0;
          auto resultBufferLength = CExports::NMH2GetID(Handle(), i, nullptr, bufferSize);
          if (resultBufferLength < 0)
            continue;
          std::vector<char> buffer(bufferSize);
          auto resultBuffer = CExports::NMH2GetID(Handle(), i, &buffer[0], bufferSize);
          if (resultBuffer < 0)
            continue;
          String keyString(buffer.begin(), buffer.end() - 1);
          nodeMaps_[keyString].Reset();
        }
        return true;
      }
 
      HandleGuard<CompositeStreamBase> handle_;
      std::atomic<Cvb::AcquisitionState> acquisitionState_;
      mutable std::map<String, Internal::AsyncRef<GenApi::NodeMap>> nodeMaps_;
      bool combinedStreaming_;
    };
    class CompositeStreamBase : public StreamBase {…};
 
  }  // namespace Driver
 
  using Driver::CompositeStreamBase;
 
  CVB_END_INLINE_NS
 
}  // namespace Cvb